使用MongoDB实现消息队列的异步消息功能_mongodb 实现消息队列

一、消息队列概述

消息队列中间件是分布式系统中重要的组件，主要解决应用耦合，异步消息，流量削锋等问题。实现高性能，高可用，可伸缩和最终一致性架构。是大型分布式系统不可缺少的中间件。

目前在生产环境，使用较多的消息队列有ActiveMQ，RabbitMQ，ZeroMQ，Kafka，MetaMQ，RocketMQ等。

自己实现一个较完善的消息队列要考虑高可用、顺序和重复消息、可靠投递、消费关系解析等等比较复杂的问题，笔者不对这些内容进行阐述，重点结合线程池实现解耦，异步消息功能，对其他功能有兴趣的话推荐美团技术博客的一篇文章消息队列设计的精髓基本都藏在本文里了

二.异步处理的流程

场景说明：用户注册后，需要发注册邮件和注册短信。传统的做法有两种
1.串行的方式；2.并行方式。
（1）串行方式：将注册信息写入数据库成功后，发送注册邮件，再发送注册短信。以上三个任务全部完成后，返回给客户端。

（2）并行方式：将注册信息写入数据库成功后，发送注册邮件的同时，发送注册短信。以上三个任务完成后，返回给客户端。与串行的差别是，并行的方式可以提高处理的时间。

假设三个业务节点每个使用50毫秒钟，不考虑网络等其他开销，则串行方式的时间是150毫秒，并行的时间可能是100毫秒。

因为CPU在单位时间内处理的请求数是一定的，假设CPU1秒内吞吐量是100次。则串行方式1秒内CPU可处理的请求量是7次（1000/150）。并行方式处理的请求量是10次（1000/100）。

小结：如以上案例描述，传统的方式系统的性能（并发量，吞吐量，响应时间）会有瓶颈。如何解决这个问题呢？

引入异步处理。改造后的架构如下：

三.使用MongoDB实现上述的架构

3.1 数据库定义

如上图所示，“发送注册短信“和“发送注册邮件“都是需要异步处理的任务。该任务将均由主线程写入数据库(“任务队列表”)，同时有另一个线程读取任务队列表的数据并根据具体的“发送注册短信“和“发送注册邮件“来处理。

任务队列模型定义
TaskQueue.java

@Document
public class TaskQueue {
   
    @Id
    private String _id;
    private String name;  //任务名称。例如：“发送注册短信“或“发送注册邮件“等
    private String param;  //需要的处理任务的参数
    private int status; // 状态，0:初始 1:处理中 8:处理失败 9:处理成功
    private int retry; // 重试次数，仅对于处理失败，status:8
    private Date created; // 创建时间
    private int threadNid; //处理成功该任务的任务处理线程唯一标示符
    private int priority; // 越小，被执行的概率越大

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

任务处理线程模型定义
ThreadInstance.java

@Document
public class ThreadInstance implements Comparable<ThreadInstance>{
   
    @Id
    private String _id;
    @Indexed(unique=true)
    private int nid; // 序号,唯一索引。唯一索引的意义后续解释
    private long pid; // 线程id
    private int taskCounts; // 当前任务处理线程正在处理的任务数，通过心跳来更新
    private int execedTasks; //该任务处理线程的总数，通过心跳来更新
    private Date update; // 上次活跃时间，通过心跳来更新
}
1
2
3
4
5
6
7
8
9
10
11
12

3.2异步处理任务的线程池

对应上图，“发送注册短信“和“发送注册邮件“等需要通过单独的线程来完成。因为可以更快的完成异步任务，可能需要多个线程同时来作为异步处理线程。此处，使用线程池来完成。
ThreadPoolExecutor线程池的工作原理

线程池有三个核心关键词：核心线程、工作队列、最大线程和饱和策略

• 核心线程池对应corePoolSize变量的值，如果运行的线程小于corePoolSize无论当前是否有空闲线程，总是会创建新的线程执行任务（这个过程需要获取全局锁）
• 如果运行的线程大于corePoolSize，则将任务加入BlockingQueue–对应工作队列
• 如果BlockingQueue已满，且当前线程数尚未超过maximumPoolSize—最大线程。则线程池继续创建线程。
• 如果BlockingQueue已满，并且当前线程数超过maximumPoolSize，则根据当前线程池已饱和。根据指定的饱和策略来处理。抛出异常，丢弃等等。

        // 创建一个顺序存储的阻塞队列，并指定大小为500
        BlockingQueue<Runnable> blockingQueue = new ArrayBlockingQueue<Runnable>(500);
        // 创建线程池的饱和策略，AbortPolicy抛异常
        RejectedExecutionHandler handler = new ThreadPoolExecutor.AbortPolicy();
        // 创建线程池，线程池基本大小5 最大线程数为10 线程最大空闲时间10分钟
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(5, 10, 10, TimeUnit.MINUTES, blockingQueue,
                handler);

        
1
2
3
4
5
6
7
8
9