性能测试中Disruptor框架shutdown失效的问题分享

FunTester

发布于 2022-04-01 09:40:55

4930

发布于 2022-04-01 09:40:55

文章被收录于专栏：FunTester

在基于Disruptor开发新的性能测试QPS模型时候，中间遇到了很多问题，踩了很多坑。今天就分享一个比较典型的问题：shutdown失效。

问题在于这么优秀的一个框架，怎么可能会存在这么明显的BUG？

经过查阅资料，还真特么存在，只不过在极少数使用场景下会发生，刚好FunTester性能测试框架设计中就属于这个场景。下面听说娓娓道来。

首先我是把每一个消费者线程都当做性能测试线程使用，此为前提。下面是两个因此带来的设定：

Disruptor框架的消费者线程或者消费者线程数组数需要在Disruptor启动之前设定，也无法修改
由于性能测试需要FunTester性能框架中基于Disruptor写的QPS模型需要设置较大的消费者或者消费者组的线程数（要达到10万QPS，这个值通常在1024以上）
性能测试QPS均为从低（多数为零）开始到设定的最大QPS
性能测试一开始，自然有大量的消费者线程处于空闲状态，甚至未启动状态

以上是四个因为Disruptor框架特性和FunTester框架设计带来的难以避免，然后就会在线程数远超（难以量化界定）需求的时候，会导致性能测试结束之后，Disruptor执行shutdown方法后，Disruptor所有线程并没有全部结束，导致程序无法正常结束且CPU使用率飙升（线程数设定较多）。具体原因大家可以自行搜索，有大佬做了非常优秀的分析、分享和演示。总结起来就是两点：

生产者的生产线程必须执行在disruptor.shutdown方法之前。
disruptor.shutdown方法必须执行在所有消费者线程启动之前。

但是这两种情况其实除非特意构造，否则极难发生，重点还是了解一点点Disruptor源码的结构和运行逻辑。经过一阵子摸索和学习，我发现了问题所在，消费者线程太多了。

在我初步的测试中，有以下几条经验：

要依旧现有数据设置消费者数量，并非越多越好
先消费者数量足够多时，QPS往往不够稳定，差异能达到30%
线程数尽量控制在2000以下，否则很容易触发Disruptor框架shutdown失效问题

PS：以上数据在QPS：5w，平均响应时间10ms设定下完成测试。

使用Disruptor做性能测试坑还是挺多的，可能之前也没人这么用过，还有几个大坑我后面会继续分享，目前总体来说，性能测试最好的模型还是线程模型，当QPS在万级别上时，QPS模型的精确很难控制。

关于较多消费者时，Disruptor框架shutdown失效的问题已经反馈给了开发者。下面是我的测试脚本，为了更容易验证，我特意写了Java版本的。


import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.TimeoutBlockingWaitStrategy;
import com.lmax.disruptor.WorkHandler;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;

import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;

public class DisJava {


    public static void main(String[] args) {
        ThreadFactory threadFactory = new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                return thread;
            }
        };
        Disruptor<Event> disruptor = new Disruptor<Event>(
                Event::new,
                256 * 256,
                threadFactory,
                ProducerType.MULTI,
                new TimeoutBlockingWaitStrategy(1000, TimeUnit.MILLISECONDS)
        );
        RingBuffer<Event> ringBuffer = disruptor.getRingBuffer();
        int num = 3000;
        EventFun[] consumers = new EventFun[num];
        for (int i = 0; i < num; i++) {
            consumers[i] = new EventFun();

        }
        disruptor.handleEventsWithWorkerPool(consumers);
        disruptor.start();
        for (int i = 0; i < 10; i++) {
            ringBuffer.publishEvent((e, s) -> {
                e.setEvent("123");
                System.out.println(System.currentTimeMillis());
            });
        }
        disruptor.shutdown();
        System.out.println("结束了");

    }

    private static class EventFun implements EventHandler<Event>, WorkHandler<Event> {

        public EventFun() {
        }

        /**
         * 多消费者
         *
         * @param event
         * @throws Exception
         */
        @Override
        public void onEvent(Event event) throws Exception {
            sleep(10);
        }

        /**
         * 单消费者
         *
         * @param event
         * @param sequence
         * @param endOfBatch
         * @throws Exception
         */
        @Override
        public void onEvent(Event event, long sequence, boolean endOfBatch) throws Exception {
            sleep(10);
        }

    }

    /**
     * 消息体
     */
    private static class Event {

        public String getEvent() {
            return event;
        }

        public void setEvent(String event) {
            this.event = event;
        }

        String event;

    }

    private static void sleep(long time) {
        try {
            Thread.sleep(time);
        } catch (InterruptedException e) {

        }
    }

}

本文参与腾讯云自媒体同步曝光计划，分享自微信公众号。

原始发表：2022-03-02，如有侵权请联系 cloudcommunity@tencent.com 删除

腾讯云测试服务