1. 消费者服务背景

   网络订单中有很多业务使用了mq,主要是为了流量高峰期业务的异步、削峰处理，提高业务的吞吐量。
    

2. 消息生产消费处理机制

   
   consumer server包含每个业务线的消息监听者。定时任务每隔1min扫描一次。

3. 线上问题产具体体现

   本次线上生产问题http://wk.mweer.com/pages/viewpage.action?pageId=9332230
   具体体现
   订单号 1081801250请求回调接口
   
   秒付服务接收到数据打印日志信息，写入消息队列
   
   消息写入队列--消息消费中间差了2分多钟。
   

4. 生产问题原因分析

   结论：消费者服务中由于integer 传null 给int导致代码问题导致消息无限重投，导致消费者线程增多并且都堆积到阻塞队列LinkBlockQueue中，当任务的执行速度小于任务的创建速度，则会出现延时的情况。

   过程：
         1、消费者在spring中的配置

   <bean id="threadPoolTaskExecutor" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor"><property name="corePoolSize" value="50"/><property name="maxPoolSize" value="300"/><property name="queueCapacity" value="1000"/><property name="threadNamePrefix" value="mf-listener-"/><property name="allowCoreThreadTimeOut" value="true"></property>
   </bean><bean id="takeawayReceiverListener" class="cn.mwee.order.listener.order.TakeawayReceiverListener"></bean>
   <bean class="org.springframework.jms.listener.SimpleMessageListenerContainer"><property name="connectionFactory" ref="jmsFactory"/><property name="messageListener" ref="takeawayReceiverListener"/><property name="concurrentConsumers" value="15"/><property name="destinationName" value="QUEUE.TAKEAWAY.TO.APPORDER"/><property name="taskExecutor" ref="threadPoolTaskExecutor"/>
   </bean>

   所有的消费者使用的是threadPoolTaskExecutor线程池，

   Set the Spring TaskExecutor to use for executing the listener once* a message has been received by the provider.

   @Override
   protected ExecutorService initializeExecutor(ThreadFactory threadFactory, RejectedExecutionHandler rejectedExecutionHandler) {BlockingQueue<Runnable> queue = createQueue(this.queueCapacity);ThreadPoolExecutor executor;if (this.taskDecorator != null) {executor = new ThreadPoolExecutor(this.corePoolSize, this.maxPoolSize, this.keepAliveSeconds, TimeUnit.SECONDS,queue, threadFactory, rejectedExecutionHandler) {@Overridepublic void execute(Runnable command) {super.execute(taskDecorator.decorate(command));}};}else {executor = new ThreadPoolExecutor(this.corePoolSize, this.maxPoolSize, this.keepAliveSeconds, TimeUnit.SECONDS,queue, threadFactory, rejectedExecutionHandler);}if (this.allowCoreThreadTimeOut) {executor.allowCoreThreadTimeOut(true);}this.threadPoolExecutor = executor;return executor;
   }/*** Create the BlockingQueue to use for the ThreadPoolExecutor.* <p>A LinkedBlockingQueue instance will be created for a positive* capacity value; a SynchronousQueue else.* @param queueCapacity the specified queue capacity* @return the BlockingQueue instance* @see java.util.concurrent.LinkedBlockingQueue* @see java.util.concurrent.SynchronousQueue*/
   protected BlockingQueue<Runnable> createQueue(int queueCapacity) {if (queueCapacity > 0) {return new LinkedBlockingQueue<Runnable>(queueCapacity);}else {return new SynchronousQueue<Runnable>();}
   }

   下面是ThreadPoolExecutor最核心的构造方法

   构造方法参数讲解 
   

   参数名	作用
   corePoolSize	核心线程池大小
   maximumPoolSize	最大线程池大小
   keepAliveTime	线程池中超过corePoolSize数目的空闲线程最大存活时间；可以allowCoreThreadTimeOut(true)使得核心线程有效时间
   TimeUnit	keepAliveTime时间单位
   workQueue	阻塞任务队列
   threadFactory	新建线程工厂
   RejectedExecutionHandler	当提交任务数超过maxmumPoolSize+workQueue之和时，任务会交给RejectedExecutionHandler来处理

   
   
   1.当线程池中超过corePoolSize线程，空闲时间达到keepAliveTime时，关闭空闲线程 
   2.当设置allowCoreThreadTimeOut(true)时，线程池中corePoolSize线程空闲时间达到keepAliveTime也将关闭 

   结合代码以及grafana分析:

   

   由于消息消费机制遇到异常会10秒重投，18号业务高峰期，从12:25-12:40开始消费者服务器的线程数从732上升到822新增了90个线程，期间就有客户反馈清台延时，而我们的消费者是共用一个线程池，核心线程为50，最大300，当线程数大于50，并且阻塞队列未满，以后会把新的消费者线程入队列等待，当任务的创建速度和处理速度差异很大，LinkedBlockingQueue会快速增涨，消费者执行也会有相应的延时。

5. 优化方案

   增加告警平台，针对消费者异常log的告警监控。
   异常消息重投机制需要优化。
   

6. 总结

   Java中integer，int转换需要注意null类型。
   消息重投需要考虑异常重试机制。
   使用线程池的地方，当服务出现异常时，重点关注线程数量变化。