1.消息重复消费问题

问题描述：
项目中启动了多个消费者,测试中发现同一条消息被多次消费。

解决方案：
①幂等方案处理
②disrutor提供了不同的处理机制：
自定义消费者实现EventHandler接口,他是属于重复消费，
自定义消费者实现WorkHandler接口,他是属于竞争消费。

重复消费：

/*** describe 消费者服务-邮件发送** @author 一叶孤舟* @date 2022年03月17日17:41*/
public class EmailEventHandler implements EventHandler {private static final Logger LOGGER = LoggerFactory.getLogger(EmailEventHandler.class);@Overridepublic void onEvent(EmailEvent event, long sequence, boolean endOfBatch) throws Exception {....}

竞争消费：

/*** describe 消费者服务-邮件发送** @author 一叶孤舟* @date 2022年03月17日17:41*/
public class EmailEventHandler implements WorkHandler {private static final Logger LOGGER = LoggerFactory.getLogger(EmailEventHandler.class);/*** 邮件发送* 根据主键更新文件发送列表数据状态** @param event 邮件发送参数*/@Overridepublic void onEvent(EmailEvent event) {...}

2.消费者线程一直处于java.lang.Thread.State: WAITING (parking)状态

问题描述：
使用了top命令（具体命令使用我在这篇文章介绍）查看测试环境服务器信息,发现CPU竟然600%。。。估计是要跑路的节奏了。。。还是说正事吧。。。使用top命令查看了服务器，发现是JAVA进程导致的，根据JAVA的进程id,top到了占用CPU的线程，jstack打印了堆栈信息,罪魁祸首就是自己写的disrutor的代码导致的，消费者线程一直parking，裂开了。认真审视了自己的代码，确实有问题，当然了这段代码不止这个问题，我太菜了。。。，已经在改正的路上了,要不然就得提桶跑路了。。。

问题代码：

/*** describe Disruptor高性能队列服务** @author 晴日朗* @date 2022年03月17日18:48*/
@Service
public class EmailSendDisruptorService implements IEmailSendDisruptorService {private static final Logger LOGGER = LoggerFactory.getLogger(EmailSendDisruptorService.class);@Autowiredprivate LookCupRepository lookCupRepository;@Overridepublic Result emailSendDisruptorData(List> mapParamList) throws Exception {ExecutorService executor = null;WorkerPool workerPool = null;try {LOGGER.info("EmailSendDisruptorService.emailSendDisruptorData.start.params={},startTime={}", JSON.toJSONString(mapParamList), DateUtils.getNowDate());if (CollectionUtils.isEmpty(mapParamList)) {return Result.errorJson(201, "发送参数为空");}// 1.创建可以缓存的线程池，提供发给consumerexecutor = ThreadPoolMonitor.newCachedThreadPool("EmailSendDisruptorServicePool");// 2.创建event工厂EmailEventFactory emailEventFactory = new EmailEventFactory();// 3.定义ringBuffer数组大小，设置为2的N次方，位运算，效率高int ringBufferSize = 1024 * 1024;// 4.创建ringBufferRingBuffer ringBuffer = RingBuffer.create(ProducerType.MULTI, emailEventFactory, ringBufferSize, new YieldingWaitStrategy());SequenceBarrier barriers = ringBuffer.newBarrier();// 5.注册消费者，可注册多个，分摊消费模式  消费者实现的是WorkHandle接口// 查询快码表,获取消费者数目,默认是3个List lookCupDOS = lookCupRepository.findAllDataByType("EMAIL_CONSUMERS");int consumersCounts = CollectionUtils.isEmpty(lookCupDOS) ? 3 : StringUtils.isEmpty(lookCupDOS.get(0).getColumn1()) ? 3 : Integer.parseInt(lookCupDOS.get(0).getColumn1());EmailEventHandler[] consumers = new EmailEventHandler[consumersCounts];for (int i = 0; i < consumers.length; i++) {consumers[i] = new EmailEventHandler();}workerPool = new WorkerPool(ringBuffer, barriers,new EventExceptionHandler(), consumers);ringBuffer.addGatingSequences(workerPool.getWorkerSequences());workerPool.start(executor);// 6.创建生产者EmailEventProducer producer = new EmailEventProducer(ringBuffer);// 7.业务参数投递for (int i = 0; i < mapParamList.size(); i++) {// 逐条发布,多线程并发处理producer.onData(mapParamList.get(i));}} catch (ParseException e) {throw new Exception(e);} finally {if (null != executor) {executor.shutdown();}}return Result.successJson(200, "操作成功", null);}
}

问题解决：

初步排查问题的根源在于配置的等待策略问题：YieldingWaitStrategy->修改为SleepingWaitStrategy。

介绍一下等待策略
①BlockingWaitStrategy
Disruptor的默认策略是BlockingWaitStrategy。在BlockingWaitStrategy内部是使用锁和condition来控制线程的唤醒。BlockingWaitStrategy是最低效的策略，但其对CPU 的消耗最小并且在各种不同部署环境中能提供更加一致的性能表现。

②SleepingWaitStrategy
SleepingWaitStrategy 的性能表现跟 BlockingWaitStrategy 差不多，对 CPU 的消耗也类似，但其对生产者线程的影响最小，通过使用LockSupport.parkNanos(1)来实现循环等待。一般来说Linux系统会暂停一个线程约60µs，这样做的好处是，生产线程不需要采取任何其他行动就可以增加适当的计数器，也不需要花费时间信号通知条件变量。但是，在 生产者线程和使用者线程之间移动事件的平均延迟会更高。它在不需要低延迟并且对生产线程的影响较小的情况最好。一个常见的用例是异步日志记录。

③YieldingWaitStrategy
YieldingWaitStrategy是可以使用在低延迟系统的策略之一。YieldingWaitStrategy 将自旋以等待序列增加到适当的值。在循环体内，将调用Thread.yield（），以允许其他排 队的线程运行。在要求极高性能且事件处理线数小于 CPU 逻辑核心数的场景中，推荐使用此策略；例如，CPU开启超线程的特性。

④BusySpinWaitStrategy
性能最好，适合用于低延迟的系统。在要求极高性能且事件处理线程数小于CPU逻辑核 心数的场景中，推荐使用此策略；例如，CPU开启超线程的特性。

3.消费者创建过多

问题描述：

我这消费者过多不是指我自定义的消费者数目太多，而是指的代码车祸现场导致消费者数目过多。。。实际场景中，多个场景注入触发这个接口，并发触发，线程池使用的是Executors.newFixedThreadPool(),创建了大量的线程（消费者），服务器卡顿，人当场裂开。。。我领导说我可以下班了。。。

问题解决：

定义复用的线程池,提供给消费者的线程复用，节省开销。

4.GC问题

问题描述：
我引入这个disruptor框架就是满足生产消费模式，需要处理的数据量很大，然后测试的过程中发现这个接口并发次数达到一定程度，页面就会卡堵。

问题解决：
①初步预计是GC影响的，后期调整VM堆分配机制，根据实际业务场景分配合理大小。
②业务测限制，只有当数据处理完了才能接受请求。

小结：

具体的调整和disruptor使用完整代码在另一篇文章：【超链接，待提供】,只要你不怕，可以看看我的代码，希望不再出现令人当场裂开的画面。