生產(chǎn)者

package cn.lonecloud.procum.disruptor;import cn.lonecloud.procum.Data; import com.lmax.disruptor.RingBuffer;import java.nio.ByteBuffer;/*** @author lonecloud* @version v1.0* @date 下午3:02 2018/5/7*/ public class Producer {//隊列private final RingBuffer<Data> dataRingBuffer;public Producer(RingBuffer<Data> dataRingBuffer) {this.dataRingBuffer = dataRingBuffer;}/*** 插入數(shù)據(jù)* @param s*/public void pushData(String s) {//獲取下一個位置long next = dataRingBuffer.next();try {//獲取容器Data data = dataRingBuffer.get(next);//設置數(shù)據(jù)data.setData(s);} finally {//插入dataRingBuffer.publish(next);}} }

　　消費者

package cn.lonecloud.procum.disruptor;import cn.lonecloud.procum.Data; import com.lmax.disruptor.WorkHandler;/*** @author lonecloud* @version v1.0* @date 下午3:01 2018/5/7*/ public class Customer implements WorkHandler<Data> {@Overridepublic void onEvent(Data data) throws Exception {System.out.println(Thread.currentThread().getName()+"---"+data.getData());} }

　　數(shù)據(jù)工廠

package cn.lonecloud.procum.disruptor;import cn.lonecloud.procum.Data; import com.lmax.disruptor.EventFactory;/*** @author lonecloud* @version v1.0* @date 下午3:02 2018/5/7*/ public class DataFactory implements EventFactory<Data> {@Overridepublic Data newInstance() {return new Data();} }

　　主函數(shù)

package cn.lonecloud.procum.disruptor;import cn.lonecloud.procum.Data; import com.lmax.disruptor.RingBuffer; import com.lmax.disruptor.dsl.Disruptor;import java.util.UUID; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors;/*** @author lonecloud* @version v1.0* @date 下午3:09 2018/5/7*/ public class Main {public static void main(String[] args) throws InterruptedException {//創(chuàng)建線程池ExecutorService service = Executors.newCachedThreadPool();//創(chuàng)建數(shù)據(jù)工廠DataFactory dataFactory = new DataFactory();//設置緩沖區(qū)大小，必須為2的指數(shù)，否則會有異常int buffersize = 1024;Disruptor<Data> dataDisruptor = new Disruptor<Data>(dataFactory, buffersize,service);//創(chuàng)建消費者線程dataDisruptor.handleEventsWithWorkerPool(new Customer(),new Customer(),new Customer(),new Customer(),new Customer(),new Customer(),new Customer());//啟動dataDisruptor.start();//獲取其隊列RingBuffer<Data> ringBuffer = dataDisruptor.getRingBuffer();for (int i = 0; i < 100; i++) {//創(chuàng)建生產(chǎn)者Producer producer = new Producer(ringBuffer);//設置內(nèi)容producer.pushData(UUID.randomUUID().toString());//Thread.sleep(1000);}} }

　　其中策略有幾種：

1. BlockingWaitStrategy:阻塞策略，最節(jié)省CPU，但是高并發(fā)條件下性能最糟糕

2 SleepingWaitStrategy:在循環(huán)中無限等待，處理數(shù)據(jù)會產(chǎn)生高延遲，對生產(chǎn)線程影響小，場景：異步日志

3. YieldingWaitStrategy:低延遲場合，使用必須保證剩余的消費者線程的邏輯CPU

4. BusySpinWaitStrategy:消費者線程會盡最大努力瘋狂的監(jiān)控緩沖區(qū)變化。

轉(zhuǎn)載于:https://www.cnblogs.com/lonecloud/p/9002927.html

總結

以上是生活随笔為你收集整理的使用Disruptor实现生产者和消费者模型的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。