当前位置：首页 > news >正文

阻塞队列以及阻塞队列的一个使用

news 2025/11/11 0:56:19

阻塞队列以及阻塞队列的一个使用

阻塞队列简介

阻塞队列（Blocking Queue）是一种常见的队列数据结构，它具有特殊的行为，可以用于多线程编程中，以协调不同线程之间的任务执行和数据传递。阻塞队列在多线程环境中非常有用，因为它可以安全地在队列为空或已满时进行线程的阻塞或唤醒操作。

阻塞队列的主要特点包括：

队列操作的线程安全性：阻塞队列提供了线程安全的队列操作，多个线程可以同时向队列中添加元素或者从队列中取出元素，而不会导致数据不一致或竞态条件的问题。
阻塞操作：当队列为空时，试图从队列中取出元素的线程会被阻塞，直到队列中有可用元素。当队列已满时，试图向队列中添加元素的线程也会被阻塞，直到队列有空间容纳新元素。
队列的大小限制：阻塞队列通常具有一个最大容量，当队列达到最大容量时，试图向队列中添加元素的线程会被阻塞。

阻塞队列的使用场景包括多线程生产者-消费者问题、线程池任务管理、事件驱动编程等，它们可以帮助有效地协调线程之间的工作，提高多线程程序的效率和可维护性。

Java中的java.util.concurrent包提供了一些常见的阻塞队列实现，如LinkedBlockingQueue和ArrayBlockingQueue等。其他编程语言和库中也有类似的实现，用于支持多线程编程。

阻塞队列的应用

阻塞队列在多线程编程中有广泛的应用，特别是在以下方面：

生产者-消费者问题：阻塞队列常用于解决生产者-消费者问题，其中生产者线程将数据放入队列，而消费者线程从队列中取出数据。阻塞队列可以有效地协调这两种操作，确保生产者和消费者之间的同步和数据的安全传递。
线程池任务管理：线程池中的任务队列通常是一个阻塞队列，等待执行的任务会被放入队列中，线程池中的工作线程从队列中取出任务并执行。这种方式可以控制同时执行的任务数量，以防止资源过度消耗。
事件驱动编程：在事件驱动编程中，事件生产者将事件放入队列，而事件消费者从队列中取出事件并响应。这种模型可以在多线程环境中实现，阻塞队列用于传递事件数据。

在Java中，你可以使用java.util.concurrent包提供的阻塞队列来方便地实现这些场景。以下是一个具体的Java示例，演示了如何使用LinkedBlockingQueue来实现一个生产者-消费者模型：

import java.util.concurrent.*;public class BlockingQueueExample {public static void main(String[] args) {// 创建一个容量为10的阻塞队列BlockingQueue<Integer> queue = new LinkedBlockingQueue<>(10);// 创建生产者线程Thread producer = new Thread(() -> {try {for (int i = 1; i <= 10; i++) {System.out.println("Producing: " + i);queue.put(i); // 将数据放入队列Thread.sleep(1000); // 模拟生产过程}} catch (InterruptedException e) {e.printStackTrace();}});// 创建消费者线程Thread consumer = new Thread(() -> {try {for (int i = 1; i <= 10; i++) {int value = queue.take(); // 从队列中取出数据System.out.println("Consuming: " + value);Thread.sleep(1500); // 模拟消费过程}} catch (InterruptedException e) {e.printStackTrace();}});// 启动生产者和消费者线程producer.start();consumer.start();}
}

在这个示例中，生产者线程不断地向阻塞队列中放入数据，而消费者线程不断地从队列中取出数据，它们之间通过阻塞队列实现了同步。这个例子演示了阻塞队列在生产者-消费者问题中的应用。

阻塞队列Springboot实现案例之线程池任务管理

有这样一个场景，如果要频繁存数据，等待数据库的响应，这样需要很长的时间，但是如果流程逻辑正确的话我们可以先把结果返回给前端，然后在异步进行数据的存储。

执行逻辑

创建一个阻塞队列：你可以选择使用java.util.concurrent包中的阻塞队列，如LinkedBlockingQueue，来存储要插入数据库的数据。也可以用ArrayBlockingQueue当作阻塞队列。

private BlockingQueue<User> blockingQueue = new ArrayBlockingQueue<>(1024);

创建一个异步方法：在Spring中，你可以使用@Async注解来定义一个异步方法。这个方法将从阻塞队列中取出数据，并将数据插入数据库。

@Async
public void processQueue() {while (true) {try {User user = blockingQueue.take(); // 从队列中取出数据userService.save(user); // 插入数据库} catch (InterruptedException e) {Thread.currentThread().interrupt();break;}}
}

配置Spring异步支持：确保你的Spring应用程序启用了异步支持。你需要在配置类中添加@EnableAsync注解，并配置一个TaskExecutor来执行异步方法。使用一个配置类

package com.xwhking.springboottemplate.config;import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;@Configuration
@EnableAsync // 启用异步支持
public class AsyncThreadPoolConfig {@Beanpublic ThreadPoolTaskExecutor taskExecutor() {ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();executor.setCorePoolSize(5); // 核心线程数executor.setMaxPoolSize(10); // 最大线程数executor.setQueueCapacity(25); // 队列容量executor.setThreadNamePrefix("MyAsyncThread-"); // 线程名称前缀executor.initialize(); // 初始化return executor;}@Beanpublic SimpleAsyncTaskExecutor simpleAsyncTaskExecutor() {return new SimpleAsyncTaskExecutor();}
}

编写Controller方法：在Controller中编写一个方法，该方法将数据放入阻塞队列。这个方法应该是同步的，因为它只是将数据放入队列，不会等待数据插入完成。

    @GetMapping("/addUser")public BaseResponse<String> testBlockQueue(String username , String password, String nickname){User user  = new User();user.setUsername(username);user.setPassword(password);user.setNickname(nickname);blockingQueue.offer(user);taskExecutor.execute(this::processQueue);return null;}

Controller完整代码

package com.xwhking.springboottemplate.controller;import com.xwhking.springboottemplate.common.BaseResponse;
import com.xwhking.springboottemplate.generator.domain.User;
import com.xwhking.springboottemplate.generator.service.UserService;
import org.springframework.core.task.TaskExecutor;
import org.springframework.scheduling.annotation.Async;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;import javax.annotation.Resource;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;@RestController
@RequestMapping("/blockQueue")
public class TestBlockQueueController {@Resourceprivate UserService userService;private BlockingQueue<User> blockingQueue = new ArrayBlockingQueue<>(1024);@ResourceTaskExecutor taskExecutor;@Asyncpublic void processQueue() {while (true) {try {User user = blockingQueue.take(); // 从队列中取出数据userService.save(user); // 插入数据库} catch (InterruptedException e) {Thread.currentThread().interrupt();break;}}}@GetMapping("/addUser")public BaseResponse<String> testBlockQueue(String username , String password, String nickname){User user  = new User();user.setUsername(username);user.setPassword(password);user.setNickname(nickname);blockingQueue.offer(user);taskExecutor.execute(this::processQueue);return null;}
}

阻塞队列以及阻塞队列的一个使用