如何使用中断器模式实现解复用器？

Question

我希望有一个环形缓冲区队列，该队列将接收对象，并以单个生产者到多个消费者的方式，将它们分布在线程池中的多个线程中。我如何使用disruptor模式来实现这一点？是否有任何HelloDemux代码示例？谢谢

Answer 1

本文详细介绍了实现中断器模式的解复用器的所有内容，但我认为线程池意味着您将需要一个调度器，这与中断器模式背道而驰。要实现demux，您需要设置固定数量的使用者线程，而不是池，并让它们从队列的尾部获取消息。现在，你可能会问，没有调度员，他们怎么能做到这一点？他们只是忙着围绕队列尾部旋转（或使用其他类型的等待策略，即旋转、让步、停车、睡眠等的组合）。现在，你可能会问，他们如何做到这一点而不相互干涉？然后有两个选项：可以使用模数（无锁）或CAS（光锁）。每一种都有自己的优缺点。模数很快，但如果一个消费者落后，则可能导致车道争用。CAS没有那么快，但不会导致车道争用。

package com.coralblocks.coralqueue.sample.demux;

import com.coralblocks.coralqueue.demux.CASAtomicDemux;
import com.coralblocks.coralqueue.demux.Demux;

public class Sample {

    private static final int NUMBER_OF_CONSUMERS = 4;

    public static void main(String[] args) throws InterruptedException {

        final Demux<StringBuilder> queue = new CASAtomicDemux<StringBuilder>(1024, StringBuilder.class, NUMBER_OF_CONSUMERS);

        Thread[] consumers = new Thread[NUMBER_OF_CONSUMERS];

        for(int i = 0; i < consumers.length; i++) {

            final int index = i;

            consumers[i] = new Thread() {

                @Override
                public void run() {

                    boolean running = true;

                    while(running) {
                        long avail;
                        while((avail = queue.availableToPoll(index)) == 0); // busy spin
                        for(int i = 0; i < avail; i++) {
                            StringBuilder sb = queue.poll(index);

                            if (sb == null) break; // mandatory for demuxes!

                            if (sb.length() == 0) {
                                running = false;
                                break; // exit immediately...
                            } else {
                                System.out.println(sb.toString());
                            }
                        }
                        queue.donePolling(index);
                    }
                }
            };

            consumers[i].start();
        }

        StringBuilder sb;

        for(int i = 0; i < 10; i++) {
            while((sb = queue.nextToDispatch()) == null); // busy spin
            sb.setLength(0);
            sb.append("message ").append(i);
            queue.flush();
        }

        // send a message to stop consumers...
        for(int i = 0; i < NUMBER_OF_CONSUMERS; i++) {
            // because the consumer exit immediately on this message, each
            // consumer will get one of these messages and exit...
            while((sb = queue.nextToDispatch()) == null); // busy spin
            sb.setLength(0);
        }
        queue.flush(); // sent batch

        for(int i = 0; i < consumers.length; i++) consumers[i].join();
    }
}