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

阻塞队列简介

阻塞队列（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();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

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

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

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

执行逻辑

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

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

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

@Async
public void processQueue() {
    while (true) {
        try {
            User user = blockingQueue.take(); // 从队列中取出数据
            userService.save(user); // 插入数据库
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            break;
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12

配置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 {
    @Bean
    public ThreadPoolTaskExecutor taskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(5); // 核心线程数
        executor.setMaxPoolSize(10); // 最大线程数
        executor.setQueueCapacity(25); // 队列容量
        executor.setThreadNamePrefix("MyAsyncThread-"); // 线程名称前缀
        executor.initialize(); // 初始化
        return executor;
    }

    @Bean
    public SimpleAsyncTaskExecutor simpleAsyncTaskExecutor() {
        return new SimpleAsyncTaskExecutor();
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

编写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;
    }
1
2
3
4
5
6
7
8
9
10

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 {
    @Resource
    private UserService userService;
    private BlockingQueue<User> blockingQueue = new ArrayBlockingQueue<>(1024);
    @Resource
    TaskExecutor taskExecutor;
    @Async
    public 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;
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

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