实现异步消费队列

无标签

发布日期: 2023-02-02

更新日期: 2023-08-27

文章字数: 595

阅读时长: 2 分

阅读次数:

参考文档：https://blog.csdn.net/u014373554/article/details/109473443

使用阻塞队列+线程池实现异步消费

1. 定义任务队列

/**
 * 定义队列
 */
private static LinkedBlockingDeque<String> deque = new LinkedBlockingDeque<>();

2. 定义线程池（单例、单线程）

线程池只可以有单个线程，这样才可以保证消费队列内容时是顺序的，多线程时会导致重复消费。
线程池需要使用常量存放，避免每次使用时创建新的线程池，而变成多个线程并发操作，也会导致重复消费。

/**
 * 消费者,单一线程， 进行处理业务逻辑
 */
public static final ExecutorService EXECUTOR_SERVICE = Executors.newSingleThreadExecutor();

3. 存储任务和顺序消费逻辑

/**
 * 模拟数据库存放数据，并保证不重复
 */
private List<String> strList = new ArrayList<>();

/**
 * 服务请求生成任务存放至队列，并调用线程消费队列
 */
@PostMapping("/process")
public void process(String equipmentNo)  &#123;
    if (StringUtils.isEmpty(equipmentNo)) &#123;
        log.info("参数不能为空");
    &#125;
    try &#123;
        if(deque.remainingCapacity() > 0)&#123;
            deque.put(equipmentNo);
        &#125;else &#123;
            log.info("队列已经满了,请稍后重试");
        &#125;

        EXECUTOR_SERVICE.submit(() ->&#123;
            //开始处理请求队列中的请求,按照队列的FIFO的规则,先处理先放入到队列中的请求
            while (deque != null && deque.size() > 0)&#123;
                //处理请求
                try &#123;
                    String orderNo = deque.take();
                    if(!strList.contains(orderNo))&#123;
                        log.info("处理订单号：&#123;&#125;",orderNo);
                        strList.add(orderNo);
                    &#125;
                &#125; catch (InterruptedException e) &#123;
                    log.info("处理订单号发生异常，异常信息：&#123;&#125;",e);
                &#125;
            &#125;
        &#125;);
    &#125; catch (InterruptedException e) &#123;
        log.info("处理发生异常，异常信息：&#123;&#125;",e);
    &#125;
&#125;

@GetMapping("/query")
public List<String> query()  &#123;
    log.info("获取strList");
    return strList;
&#125;