用Disruptor框架实现生产者-消费者模式

本文主要是介绍用Disruptor框架实现生产者-消费者模式，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

ConcurrentLinkedQueue队列的秘诀就在于大量使用了无锁CAS操作。
现成的Disruptor框架实现CAS进行编程。
无锁的缓存框架：Disruptor
它使用无锁的方式实现了一个环形队列，非常适合实现生产者-消费者模式，
比如事件和消息的发布。如果队列是环形的，则只需要对外提供一个当前位置cursor,
利用这个指针即可用进入队操作，也可用进行出队操作。
由于环形队列的缘故，队列的总大小必须事先指定，不能动态扩展。
为了能够快速从一个序列对于到数组的实际位置，每次有元素入队，序列就加1.
Disruptor框架要求我们必须将数组的大小设置为2的整数次方。这样通过sequence&(queueSize-1)
就能立即定位到实际的元素位置index,这比取余%操作快得多。
如果搭建不理解上面的sequence&(queueSize-1),那么我在这里再简单说明一下。
如果queueSize是2的整数次幂。则这个数字的二进制表示必然是10、100、1000、10000等形式。

果一个数字是2的整数次幂，其二进制表示的确是形如10、100、1000、10000等的形式。
这是因为2的整数次幂在二进制中只有一个比特位为1，其余都为0。例如：
2^1 = 2，二进制表示为10
2^2 = 4，二进制表示为100
2^3 = 8，二进制表示为1000
2^4 = 16，二进制表示为10000
这种规律一直持续下去。这样的性质在计算机科学和计算机工程中经常会被利用，
特别是在处理队列（Queue）等数据结构的大小时。
Disruptor框架

生产者需要一个 RingBuffer的引用，也就是环形缓冲区。
它有一个重要的方法pushData()将产生的数据推入缓冲区。
方法pushData()接收一个ByteBuffer对象。在ByteBuffer对象中
可用用来包装任何数据类型。这里用来存储long整数，
pushData()方法的功能就是将传入的ByteBuffer对象中的数据提取出来，
并转载到环形缓冲区中。
只有发布后的数据才会真正被消费者看见。

public class Consumer implements WorkHandler<PCData> {@Overridepublic void onEvent(PCData event) throws Exception{System.out.println(Thread.currentThread().getId()+":Event: --"+ event.get() * event.get() +"--");}}

public class PCData {private long value;public void set(long value){this.value = value;}public long get(){return value;}}

public class PCDataFactory implements EventFactory<PCData> {public PCData newInstance(){return new PCData();}
}

public class Producer {private final RingBuffer<PCData> ringBuffer;public Producer(RingBuffer<PCData> ringBuffer){this.ringBuffer = ringBuffer;}public void pushData(ByteBuffer bb){long sequence = ringBuffer.next();try{PCData event = ringBuffer.get(sequence);event.set(bb.getLong(0));}finally {ringBuffer.publish(sequence);}}}

public static void main(String[] args) {Executor executors =  Executors.newCachedThreadPool();PCDataFactory factory = new PCDataFactory();int bufferSize = 1024;Disruptor<PCData> disruptor = new Disruptor<PCData>((EventFactory<PCData>) factory,bufferSize,(Executor) executors,ProducerType.MULTI,new BlockingWaitStrategy());disruptor.handleEventsWithWorkerPool(new Consumer(),new Consumer(),new Consumer(),new Consumer());disruptor.start();RingBuffer<PCData> ringBuffer = disruptor.getRingBuffer();Producer producer = new Producer(ringBuffer);ByteBuffer bb = ByteBuffer.allocate(8);for(long l = 0;true;l++){bb.putLong(0,l);producer.pushData(bb);try {Thread.sleep(100);} catch (InterruptedException e) {throw new RuntimeException(e);}System.out.println("add data "+ l);}}

main 函数是一个无限循环，因为它包含了一个 for 循环，其中的条件是 true。因此，这个 main 函数会一直执行下去，直到程序被手动中断或出现异常导致程序终止。

在循环中，你不断地向 RingBuffer 中发布数据，由 Producer 类的 pushData 方法完成。同时，RingBuffer 中的数据会被多个消费者（Consumer 类）并发地处理。

你在循环中使用了 Thread.sleep(100)，这会导致每次循环执行后线程暂停 100 毫秒。因此，每次数据被发布后，你会看到 "add data" 的输出，并且在消费者的 onEvent 方法中会输出相应的信息。

由于这是一个生产者-消费者模型，生产者生产数据，消费者处理数据，程序在生产和消费之间持续运行。

如果你希望在某个条件下结束程序，你需要在循环中添加相应的退出条件，并在满足条件时使用 break 语句跳出循环。否则，这个程序将一直执行下去。