使用线程池+内存队列实现异步处理业务问题

image-20191201143250725

开源中国地址 https://my.oschina.net/jiansin

背景

当系统中的业务存在大量的相同任务（比如发送大量邮件），并且每个任务花费的时间也比较长，前段需要较快 的响应，针对这种需求，我们可以采用消息队列进行异步通知，同时也可以采用线程池+内存队列实现异步通知，处理业务问题。

代码实现

以下采用发送邮件作为demo

邮箱实体类

@Data
public class Email implements Serializable {
    
    private static final long serialVersionUID = 1L;
    
    /**
     * 自增主键
     */
    private Long id;
    
    /**
     * 接收人邮箱(多个逗号分开)
     */
    private String receiveEmail;
    
    /**
     * 主题
     */
    private String subject;
    
    /**
     * 发送内容
     */
    private String content;
    
    /**
     * 模板
     */
    private String template;
    
    /**
     * 发送时间
     */
    private Timestamp sendTime;
}

邮件队列

public class MailQueue {
     //队列大小
    static final int QUEUE_MAX_SIZE   = 1000;

    static BlockingQueue<Email> blockingQueue = new LinkedBlockingQueue<Email>(QUEUE_MAX_SIZE);
    
    /**
     * 私有的默认构造子，保证外界无法直接实例化
     */
    private MailQueue(){};
    /**
     * 类级的内部类，也就是静态的成员式内部类，该内部类的实例与外部类的实例
     * 没有绑定关系，而且只有被调用到才会装载，从而实现了延迟加载
     */
    private static class SingletonHolder{
        /**
         * 静态初始化器，由JVM来保证线程安全
         */
            private  static MailQueue queue = new MailQueue();
    }
    //单例队列
    public static MailQueue getMailQueue(){
        return SingletonHolder.queue;
    }
    //生产入队
    public  void  produce(Email mail) throws InterruptedException {
        blockingQueue.put(mail);
    }
    //消费出队
    public  Email consume() throws InterruptedException {
        return blockingQueue.take();
    }
    // 获取队列大小
    public int size() {
        return blockingQueue.size();
    }
}

邮件消费队列 实际上一次1000的并发请求最终的有效转化数为2 =CORE_POOL_SIZE+WORK_QUEUE_SIZE，

当CORE_POOL_SIZE=2，MAX_POOL_SIZE=5，WORK_QUEUE_SIZE=50时，1000的并发请求有效转化数为55

个，所以自己可以根据自己的业务访问量设置队列缓冲池的大小和最大的线程数量。

@Component
public class ConsumeMailQueue {
    private static final Logger logger = LoggerFactory.getLogger(ConsumeMailQueue.class);

  //普通的业务类
    @Autowired
    IMailService mailService;

    // 线程池维护线程的最少数量
    private final static int CORE_POOL_SIZE = 1;
    // 线程池维护线程的最大数量
    private final static int MAX_POOL_SIZE = 1;
    // 线程池维护线程所允许的空闲时间
    private final static int KEEP_ALIVE_TIME = 0;
    // 线程池所使用的缓冲队列大小
    private final static int WORK_QUEUE_SIZE = 1;

    // 消息缓冲队列
    Queue<Email> msgQueue = new LinkedList<Email>();


    //由于超出线程范围和队列容量而使执行被阻塞时所使用的处理程序
    final RejectedExecutionHandler handler = new RejectedExecutionHandler() {
        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
            logger.info("线程池太忙了处理不了过多任务.........多余的线程将会放入msgQueue");
            //可以新开调度器进行处理这些调度任务，或者把没处理的任务保存到数据库中，然后定时任务继续处理
            msgQueue.add(((PollMail)r).getEmail());


        }
    };

    // 任务线程池
    final ThreadPoolExecutor threadPool = new ThreadPoolExecutor(
            CORE_POOL_SIZE, MAX_POOL_SIZE, KEEP_ALIVE_TIME,
            TimeUnit.SECONDS, new ArrayBlockingQueue(WORK_QUEUE_SIZE), handler);


    // 调度线程池。此线程池支持定时以及周期性执行任务的需求。
    final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);


    @PostConstruct
    public void init() {

        //开启邮件消费队列检查
        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    while (true){
                        Email mail = MailQueue.getMailQueue().consume();
                        logger.info("剩余邮件总数:{}",MailQueue.getMailQueue().size());
                        threadPool.execute(new PollMail(mailService,mail));
                    }
                } catch (InterruptedException e) {
                    logger.info("邮件队列消费失败，失败原因为---->",e.getMessage());
                }


            }
        }).start();
    }


    //lombok
    @Data
    class PollMail implements Runnable {

        IMailService mailService;

        Email email;

        public PollMail(IMailService mailService,Email email) {
            this.mailService = mailService;
            this.email = email;
        }

        @Override
        public void run() {
            logger.info("正在处理的邮件为----->{}",this.email.getEmail());
            mailService.dealSend(this.email);
        }
    }
    @PreDestroy
    public void stopThread() throws InterruptedException {

        /**
         * pool.awaitTermination(1, TimeUnit.SECONDS)
         * 会每隔一秒钟检查一次是否执行完毕（状态为 TERMINATED），
         * 当从 while 循环退出时就表明线程池已经完全终止了。
         */
        scheduler.shutdown();
        threadPool.shutdown();
        while (!threadPool.awaitTermination(1, TimeUnit.SECONDS)) {
            logger.info("线程还在执行。。。");
        }
    }


}

        

控制层代码如下

@Api(tags = "邮件管理")
@RestController
@RequestMapping("/mail")
public class mailController {

    @Autowired
    private IMailService mailService;

    @PostMapping("send")
    public Result send(Email mail) throws InterruptedException {
        mailService.send(mail);
        return Result.ok();
    }

}

接口层代码如下

/**
 * @description:
 * @author: 简单的心
 * @version:
 * @modified By:1170370113@qq.com
 */
public interface IMailService {

    /**
     * 邮件发送业务
     * @param mail
     */
    void send(Email mail) throws InterruptedException;

    /**
     * 处理需要发送的邮件任务
     * @param email
     */
    void dealSend(Email email);
}

接口实现层代码如下

@Service
public class MailServiceImpl implements IMailService {
    private static final Logger logger = LoggerFactory.getLogger(MailServiceImpl.class);


    @Override
    public void send(Email mail) {

        try {
            MailQueue.getMailQueue().produce(mail);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    //用于统计达到的有效请求
    AtomicInteger flag=new AtomicInteger(0);

    @Override
    public void dealSend(Email email) {
        try {
            Thread.sleep(500);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        logger.info("邮件信息已经发送，具体内容如------->{}",email.toString());
        logger.info("总共达到的有效请求 {}",flag.addAndGet(1));
    }

}