阻塞队列实现异步消费

业务场景

• 通过极光推送向设备端发送消息通知，为了用户能够得到及时响应，采用异步消息推送。将用户的请求发送给消息队列，用户实时响应，启用一个线程消费，发送消息。

核心代码

// 缓存推送消息
private BlockingQueue<SendPushParam> sendPushQueue = new LinkedBlockingQueue<>(2000);

// 消费者线程：servlet启动时加载,启动这个线程发送推送指令
@PostConstruct
public void initQueueConsumer() {
    
    log.debug("创建队列消费线程");

    this.consumerThread = new Thread(() -> {
    
        while (true) {
    
            try {
    
                SendPushParam sendPushParam = sendPushQueue.take();
                if (sendPushParam == null) {
    
                    continue;
                }
                log.trace("消费推送请求信息: {}", sendPushParam);
                sendPushDirectly(sendPushParam);
            } catch (Exception e) {
    
                log.error("消费推送请求信息失败: {}", e.getMessage());
            }
        }
    });

    this.consumerThread.start();
}

// 生产者线程：发送推送则写入消息队列
@Override
public void sendPush(SendPushParam param) {
    
    if (Boolean.TRUE.equals(param.getAsync())) {
    
        boolean offered = sendPushQueue.offer(param);
        if (!offered) {
    
            log.error("异步推送, 加入队列失败，尝试直接推送");
        }
        log.trace("异步推送加入队列成功");
    } else {
    
        log.debug("直接推送");
        sendPushDirectly(param);
    }
}

参考阅读

1. BlockingQueue
2. [BlockingQueue（阻塞队列）](