SpringBoot Disruptor框架遇到的问题

Posted 2022-12-26 地表最强菜鸡

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了SpringBoot Disruptor框架遇到的问题相关的知识，希望对你有一定的参考价值。

1.消息重复消费问题

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

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

重复消费：

/**
 * describe 消费者服务-邮件发送
 *
 * @author 一叶孤舟
 * @date 2022年03月17日17:41
 */
public class EmailEventHandler implements EventHandler<EmailEvent> 

    private static final Logger LOGGER = LoggerFactory.getLogger(EmailEventHandler.class);

    @Override
    public void onEvent(EmailEvent event, long sequence, boolean endOfBatch) throws Exception 
        ....

竞争消费：

/**
 * describe 消费者服务-邮件发送
 *
 * @author 一叶孤舟
 * @date 2022年03月17日17:41
 */
public class EmailEventHandler implements WorkHandler<EmailEvent> 

    private static final Logger LOGGER = LoggerFactory.getLogger(EmailEventHandler.class);

    /**
     * 邮件发送
     * 根据主键更新文件发送列表数据状态
     *
     * @param event 邮件发送参数
     */
    @Override
    public 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);

    @Autowired
    private LookCupRepository lookCupRepository;

    @Override
    public Result emailSendDisruptorData(List<Map<String, Object>> mapParamList) throws Exception 
        ExecutorService executor = null;
        WorkerPool<EmailEvent> workerPool = null;

        try 
            LOGGER.info("EmailSendDisruptorService.emailSendDisruptorData.start.params=,startTime=", JSON.toJSONString(mapParamList), DateUtils.getNowDate());
            if (CollectionUtils.isEmpty(mapParamList)) 
                return Result.errorJson(201, "发送参数为空");
            
            // 1.创建可以缓存的线程池，提供发给consumer
            executor = ThreadPoolMonitor.newCachedThreadPool("EmailSendDisruptorServicePool");
            // 2.创建event工厂
            EmailEventFactory emailEventFactory = new EmailEventFactory();
            // 3.定义ringBuffer数组大小，设置为2的N次方，位运算，效率高
            int ringBufferSize = 1024 * 1024;
            // 4.创建ringBuffer
            RingBuffer<EmailEvent> ringBuffer = RingBuffer.create(ProducerType.MULTI, emailEventFactory, ringBufferSize, new YieldingWaitStrategy());
            SequenceBarrier barriers = ringBuffer.newBarrier();
            // 5.注册消费者，可注册多个，分摊消费模式  消费者实现的是WorkHandle接口
            // 查询快码表,获取消费者数目,默认是3个
            List<LookCupDO> 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<EmailEvent>(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开启超线程的特性。