0

我知道我的问题违背了 Disruptor API 的基本主张。但是当我了解它时，我编写了一个程序来替换我使用 ArrayLinkedBlockingQueue 的 1P-1C 用例。但是当我运行程序时，我一直在使用中断器获得比 ArrayLinkedBlockingQueue 更差的总时间。我一定是做错了什么或测量错了，但我不确定它在我的程序中是什么。有人有意见吗？

（这是一个测试程序，所以显然我的 EventHandler 没有做任何事情）

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.lmax.disruptor.BusySpinWaitStrategy;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.EventTranslator;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;

public class SPSCDisruptorTest {
    private static final int UNIT_SIZE = 1024;
    private static final int BUFFER_SIZE = UNIT_SIZE * 1024 * 16;
    private static final int ITERATIONS = BUFFER_SIZE;
    private static final Logger logger = LoggerFactory
            .getLogger(SPSCDisruptorTest.class);

    private static class Data {
        private String data;

        public String getData() {
            return data;
        }

        public void setData(String data) {
            this.data = data;
        }

        @Override
        public String toString() {
            return "Data [data=" + data + "]";
        }

        public final static EventFactory<Data> DATA_FACTORY = new EventFactory<Data>() {

            @Override
            public Data newInstance() {
                return new Data();
            }

        };
    }

    private static class DataEventTranslator implements EventTranslator<Data> {
        private String payload;

        public DataEventTranslator(String payload) {
            this.payload = payload;
        }

        @Override
        public void translateTo(Data d, long sequence) {
            d.setData(payload);
        }

    };

    public static void main(String[] args) throws InterruptedException {
        new SPSCDisruptorTest().testDisruptor();
        new SPSCDisruptorTest().testExecutor();
    }

    @SuppressWarnings("unchecked")
    public void testDisruptor() {
        ExecutorService exec = Executors.newSingleThreadExecutor();
        Disruptor<Data> disruptor = new Disruptor<Data>(
                SPSCDisruptorTest.Data.DATA_FACTORY, BUFFER_SIZE, exec,
                ProducerType.SINGLE, new BusySpinWaitStrategy());
        disruptor.handleEventsWith(new EventHandler<Data>() {

            @Override
            public void onEvent(Data data, long sequence, boolean endOfBatch)
                    throws Exception {
            }

        });
        long t1 = System.nanoTime();
        RingBuffer<Data> buffer = disruptor.start();
        for (int i = 1; i <= ITERATIONS; i++) {
            buffer.publishEvent(new DataEventTranslator("data" + i));
        }
        logger.info("waiting for shutdown");
        disruptor.shutdown();
        logger.info("Disruptor Time (ms): " + (System.nanoTime() - t1 * 1.0)
                / 1000);
        logger.info("Disruptor is shutdown");
        exec.shutdown();
    }

    public void testExecutor() throws InterruptedException {
        ExecutorService executor = new ThreadPoolExecutor(1, 1, 0L,
                TimeUnit.MILLISECONDS, new ArrayBlockingQueue<Runnable>(
                        BUFFER_SIZE));
        long t1 = System.nanoTime();
        for (int i = 1; i <= ITERATIONS; i++) {
            executor.submit(new DataRunner("data" + i));
        }
        executor.shutdown();
        executor.awaitTermination(5000, TimeUnit.SECONDS);
        logger.info("Executor Time (ms): " + (System.nanoTime() - t1 * 1.0)
                / 1000);
    }

    private static class DataRunner implements Runnable {
        private String data;

        public DataRunner(String data) {
            this.data = data;
        }

        @Override
        public void run() {
        }

    }
}
4

2 回答 2

0

您实际上是在测量错误。您应该在启动干扰器后开始测量，因为预热需要时间（分配环形缓冲区）。由于您的缓冲区大小很大，因此预热需要相当长的时间。试试下面的示例代码。它应该给你更好的时间。

    RingBuffer<Data> buffer = disruptor.start();
    long t1 = System.nanoTime();
    for (int i = 1; i <= ITERATIONS; i++) {
        buffer.publishEvent(new DataEventTranslator("data" + i));
    }
    logger.info("waiting for shutdown");
    disruptor.shutdown();
    logger.info("Disruptor Time (ms): " + (System.nanoTime() - t1 * 1.0)
            / 1000);
于 2014-02-06T12:42:59.547 回答
0

您没有足够的争论来展示无锁破坏者如何提供帮助。特别是，您的执行程序队列与迭代一样大！所有数据都适合执行程序队列，因此它基本上不会在 notempty/notfull 条件下运行。

执行器服务也很糟糕，因为如果队列更小，您将拒绝执行。您需要比较的是具有有限队列（可能有 1000 长）和阻塞 .put()/.take() 调用的 2 个线程。

更糟糕的是，您需要成批的数据（不是 1 对 1）和许多读者，甚至可能是许多作者。在 executor 测试中使用竞争队列访问，disruptor 显示其性能应该没有问题。

于 2018-04-17T23:14:45.877 回答