先c1和c2并行消费生产者产生的数据，然后c3再消费该数据

我们来使用代码实现：我们可以使用Disruptor实例来实现，也可以不用产生Disruptor实例，直接调用RingBuffer的api来实现，不清楚看上一节使用的api函数workpool和BatchEventProcess来辅助实现消费者

上面需要使用的场景很复杂，只能使用Disruptor实例来实现线程通信，简单场景就直接使用RingBuffer就可以了

我们来看下程序的代码：

案例一：

package bhz.generate2;

import java.util.UUID;

import bhz.generate1.Trade;

import com.lmax.disruptor.EventHandler;

import com.lmax.disruptor.WorkHandler;

public class Handler1 implements EventHandler<Trade>,WorkHandler<Trade> {  

    @Override

    public void onEvent(Trade event, long sequence, boolean endOfBatch) throws Exception {

        this.onEvent(event);

    }  

    @Override

    public void onEvent(Trade event) throws Exception {

        System.out.println("handler1: set name");

        event.setName("h1");

        Thread.sleep(1000);

    }

}  
package bhz.generate2;

import bhz.generate1.Trade;

import com.lmax.disruptor.EventHandler;

public class Handler2 implements EventHandler<Trade> {  

    @Override

    public void onEvent(Trade event, long sequence,  boolean endOfBatch) throws Exception {

        System.out.println("handler2: set price");

        event.setPrice(17.0);

        Thread.sleep(1000);

    }  

}  
package bhz.generate2;

import bhz.generate1.Trade;

import com.lmax.disruptor.EventHandler;

public class Handler3 implements EventHandler<Trade> {

    @Override

    public void onEvent(Trade event, long sequence,  boolean endOfBatch) throws Exception {

        System.out.println("handler3: name: " + event.getName() + " , price: " + event.getPrice() + ";  instance: " + event.toString());

    }

}

我们来看下主函数

package bhz.generate2;

import java.util.concurrent.CountDownLatch;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

import bhz.generate1.Trade;

import bhz.generate1.TradeHandler;

import com.lmax.disruptor.BusySpinWaitStrategy;

import com.lmax.disruptor.EventFactory;

import com.lmax.disruptor.dsl.Disruptor;

import com.lmax.disruptor.dsl.EventHandlerGroup;

import com.lmax.disruptor.dsl.ProducerType;

public class Main {

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

        long beginTime=System.currentTimeMillis();

        int bufferSize=1024;

        ExecutorService executor=Executors.newFixedThreadPool(8);  

        Disruptor<Trade> disruptor = new Disruptor<Trade>(new EventFactory<Trade>() {

            @Override

            public Trade newInstance() {

                return new Trade();

            }

        }, bufferSize, executor, ProducerType.SINGLE, new BusySpinWaitStrategy());  

        //菱形操作

        //使用disruptor创建消费者组C1,C2

        EventHandlerGroup<Trade> handlerGroup =

                disruptor.handleEventsWith(new Handler1(), new Handler2());

        //声明在C1,C2完事之后执行JMS消息发送操作 也就是流程走到C3

        handlerGroup.then(new Handler3());

        //顺序操作

        /**

        disruptor.handleEventsWith(new Handler1()).

            handleEventsWith(new Handler2()).

            handleEventsWith(new Handler3());

        */

        //六边形操作.

        /**

        Handler1 h1 = new Handler1();

        Handler2 h2 = new Handler2();

        Handler3 h3 = new Handler3();

        Handler4 h4 = new Handler4();

        Handler5 h5 = new Handler5();

        disruptor.handleEventsWith(h1, h2);

        disruptor.after(h1).handleEventsWith(h4);

        disruptor.after(h2).handleEventsWith(h5);

        disruptor.after(h4, h5).handleEventsWith(h3);

        */

        disruptor.start();//启动

        CountDownLatch latch=new CountDownLatch(1);

        //生产者准备

        executor.submit(new TradePublisher(latch, disruptor));

        latch.await();//等待生产者完事. 

        disruptor.shutdown();

        executor.shutdown();

        System.out.println("总耗时:"+(System.currentTimeMillis()-beginTime));

    }

}  

生产者的代码：
package bhz.generate2;

import java.util.Random;

import java.util.concurrent.CountDownLatch;

import bhz.generate1.Trade;

import com.lmax.disruptor.EventTranslator;

import com.lmax.disruptor.dsl.Disruptor;

public class TradePublisher implements Runnable {  

    Disruptor<Trade> disruptor;

    private CountDownLatch latch;  

    private static int LOOP=10;//模拟百万次交易的发生  

    public TradePublisher(CountDownLatch latch,Disruptor<Trade> disruptor) {

        this.disruptor=disruptor;

        this.latch=latch;

    }  

    @Override

    public void run() {

        TradeEventTranslator tradeTransloator = new TradeEventTranslator();

        for(int i=0;i<LOOP;i++){

            disruptor.publishEvent(tradeTransloator);

        }

        latch.countDown();

    }  

}  

class TradeEventTranslator implements EventTranslator<Trade>{  

    private Random random=new Random();  

    @Override

    public void translateTo(Trade event, long sequence) {

        this.generateTrade(event);

    }  

    private Trade generateTrade(Trade trade){

        trade.setPrice(random.nextDouble()*9999);

        return trade;

    }  

}  


hanlde1设置name，handle2设置价格，hanle3将名字和价格打印出来


我们来看下程序运行的代码：
 


handler2: set price
handler1: set name
handler2: set price
handler1: set name
handler3: name: h1 , price: 17.0;  instance: bhz.generate1.Trade@2aa2f9e6
handler2: set price
handler1: set name
handler2: set price
handler1: set name
handler1: set name
handler2: set price
handler1: set name
handler2: set price
handler2: set price
handler1: set name
handler2: set price
handler1: set name
handler1: set name
handler2: set price
handler1: set name
handler2: set price
handler3: name: h1 , price: 17.0;  instance: bhz.generate1.Trade@7d6c848f
handler3: name: h1 , price: 17.0;  instance: bhz.generate1.Trade@5f73089d
handler3: name: h1 , price: 17.0;  instance: bhz.generate1.Trade@793aac5f
handler3: name: h1 , price: 17.0;  instance: bhz.generate1.Trade@7b0acf26
handler3: name: h1 , price: 17.0;  instance: bhz.generate1.Trade@2a606e6
handler3: name: h1 , price: 17.0;  instance: bhz.generate1.Trade@620ee765
handler3: name: h1 , price: 17.0;  instance: bhz.generate1.Trade@4079ca2e
handler3: name: h1 , price: 17.0;  instance: bhz.generate1.Trade@7bc8b313
handler3: name: h1 , price: 17.0;  instance: bhz.generate1.Trade@3564e3e2
总耗时:10027


上面对应一个生产者，三个消费者，一个生产者提交了10个任务，上面每一个hashcode都是不同的


上面这种模式是一个生产者，多个消费者的情况


场景2：








我们来看下程序的代码：




package bhz.generate2;

import java.util.UUID;

import bhz.generate1.Trade;

import com.lmax.disruptor.EventHandler;

import com.lmax.disruptor.WorkHandler;

public class Handler4 implements EventHandler<Trade>,WorkHandler<Trade> {  

    @Override

    public void onEvent(Trade event, long sequence, boolean endOfBatch) throws Exception {

        this.onEvent(event);

    }  

    @Override

    public void onEvent(Trade event) throws Exception {

        System.out.println("handler4: get name : " + event.getName());

        event.setName(event.getName() + "h4");

    }

}  






package bhz.generate2;

import java.util.UUID;

import bhz.generate1.Trade;

import com.lmax.disruptor.EventHandler;

import com.lmax.disruptor.WorkHandler;

public class Handler5 implements EventHandler<Trade>,WorkHandler<Trade> {  

    @Override

    public void onEvent(Trade event, long sequence, boolean endOfBatch) throws Exception {

        this.onEvent(event);

    }  

    @Override

    public void onEvent(Trade event) throws Exception {

        System.out.println("handler5: get price : " + event.getPrice());

        event.setPrice(event.getPrice() + 3.0);

    }

}  




主函数：




package bhz.generate2;

import java.util.concurrent.CountDownLatch;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

import bhz.generate1.Trade;

import bhz.generate1.TradeHandler;

import com.lmax.disruptor.BusySpinWaitStrategy;

import com.lmax.disruptor.EventFactory;

import com.lmax.disruptor.dsl.Disruptor;

import com.lmax.disruptor.dsl.EventHandlerGroup;

import com.lmax.disruptor.dsl.ProducerType;

public class Main {

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

        long beginTime=System.currentTimeMillis();

        int bufferSize=1024;

        ExecutorService executor=Executors.newFixedThreadPool(8);  

        Disruptor<Trade> disruptor = new Disruptor<Trade>(new EventFactory<Trade>() {

            @Override

            public Trade newInstance() {

                return new Trade();

            }

        }, bufferSize, executor, ProducerType.SINGLE, new BusySpinWaitStrategy());  

  /*      //菱形操作

        //使用disruptor创建消费者组C1,C2

        EventHandlerGroup<Trade> handlerGroup =

                disruptor.handleEventsWith(new Handler1(), new Handler2());

        //声明在C1,C2完事之后执行JMS消息发送操作 也就是流程走到C3

        handlerGroup.then(new Handler3());*/

        //顺序操作

        /**

        disruptor.handleEventsWith(new Handler1()).

            handleEventsWith(new Handler2()).

            handleEventsWith(new Handler3());

        */

        //六边形操作. 

        Handler1 h1 = new Handler1();

        Handler2 h2 = new Handler2();

        Handler3 h3 = new Handler3();

        Handler4 h4 = new Handler4();

        Handler5 h5 = new Handler5();

        disruptor.handleEventsWith(h1, h2);

        disruptor.after(h1).handleEventsWith(h4);

        disruptor.after(h2).handleEventsWith(h5);

        disruptor.after(h4, h5).handleEventsWith(h3);

        disruptor.start();//启动

        CountDownLatch latch=new CountDownLatch(1);

        //生产者准备

        executor.submit(new TradePublisher(latch, disruptor));

        latch.await();//等待生产者完事. 

        disruptor.shutdown();

        executor.shutdown();

        System.out.println("总耗时:"+(System.currentTimeMillis()-beginTime));

    }

}  




程序运行的效果：


handler2: set price
handler1: set name
handler2: set price
handler1: set name
handler2: set price
handler1: set name
handler2: set price
handler1: set name
handler1: set name
handler2: set price
handler1: set name
handler2: set price
handler1: set name
handler2: set price
handler1: set name
handler2: set price
handler1: set name
handler2: set price
handler2: set price
handler1: set name
handler5: get price : 17.0
handler5: get price : 17.0
handler5: get price : 17.0
handler5: get price : 17.0
handler5: get price : 17.0
handler5: get price : 17.0
handler5: get price : 17.0
handler5: get price : 17.0
handler5: get price : 17.0
handler5: get price : 17.0
handler4: get name : h1
handler4: get name : h1
handler4: get name : h1
handler4: get name : h1
handler4: get name : h1
handler4: get name : h1
handler4: get name : h1
handler4: get name : h1
handler4: get name : h1
handler4: get name : h1
handler3: name: h1h4 , price: 20.0;  instance: bhz.generate1.Trade@2f30e44c
handler3: name: h1h4 , price: 20.0;  instance: bhz.generate1.Trade@32dc51c8
handler3: name: h1h4 , price: 20.0;  instance: bhz.generate1.Trade@1e7d3b8e
handler3: name: h1h4 , price: 20.0;  instance: bhz.generate1.Trade@2fd0f745
handler3: name: h1h4 , price: 20.0;  instance: bhz.generate1.Trade@4019eb89
handler3: name: h1h4 , price: 20.0;  instance: bhz.generate1.Trade@65b6903f
handler3: name: h1h4 , price: 20.0;  instance: bhz.generate1.Trade@34b84c44
handler3: name: h1h4 , price: 20.0;  instance: bhz.generate1.Trade@2f971f66
handler3: name: h1h4 , price: 20.0;  instance: bhz.generate1.Trade@6a8e79f2
handler3: name: h1h4 , price: 20.0;  instance: bhz.generate1.Trade@2fe83585
总耗时:10368


3、案例三：如何实现按顺利操作了，c1执行完了执行c2，c2执行完了执行c3




package bhz.generate2;

import java.util.concurrent.CountDownLatch;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

import bhz.generate1.Trade;

import bhz.generate1.TradeHandler;

import com.lmax.disruptor.BusySpinWaitStrategy;

import com.lmax.disruptor.EventFactory;

import com.lmax.disruptor.dsl.Disruptor;

import com.lmax.disruptor.dsl.EventHandlerGroup;

import com.lmax.disruptor.dsl.ProducerType;

public class Main {

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

        long beginTime=System.currentTimeMillis();

        int bufferSize=1024;

        ExecutorService executor=Executors.newFixedThreadPool(8);  

        Disruptor<Trade> disruptor = new Disruptor<Trade>(new EventFactory<Trade>() {

            @Override

            public Trade newInstance() {

                return new Trade();

            }

        }, bufferSize, executor, ProducerType.SINGLE, new BusySpinWaitStrategy());  

  /*      //菱形操作

        //使用disruptor创建消费者组C1,C2

        EventHandlerGroup<Trade> handlerGroup =

                disruptor.handleEventsWith(new Handler1(), new Handler2());

        //声明在C1,C2完事之后执行JMS消息发送操作 也就是流程走到C3

        handlerGroup.then(new Handler3());*/

        //顺序操作

        disruptor.handleEventsWith(new Handler1()).

            handleEventsWith(new Handler2()).

            handleEventsWith(new Handler3());

        //六边形操作.

    /*

        Handler1 h1 = new Handler1();

        Handler2 h2 = new Handler2();

        Handler3 h3 = new Handler3();

        Handler4 h4 = new Handler4();

        Handler5 h5 = new Handler5();

        disruptor.handleEventsWith(h1, h2);

        disruptor.after(h1).handleEventsWith(h4);

        disruptor.after(h2).handleEventsWith(h5);

        disruptor.after(h4, h5).handleEventsWith(h3);*/

        disruptor.start();//启动

        CountDownLatch latch=new CountDownLatch(1);

        //生产者准备

        executor.submit(new TradePublisher(latch, disruptor));

        latch.await();//等待生产者完事. 

        disruptor.shutdown();

        executor.shutdown();

        System.out.println("总耗时:"+(System.currentTimeMillis()-beginTime));

    }

}  




执行执行的效果如下所示：


handler1: set name
handler1: set name
handler1: set name
handler1: set name
handler1: set name
handler1: set name
handler1: set name
handler1: set name
handler1: set name
handler1: set name
handler2: set price
handler2: set price
handler2: set price
handler2: set price
handler2: set price
handler2: set price
handler2: set price
handler2: set price
handler2: set price
handler2: set price
handler3: name: h1 , price: 17.0;  instance: bhz.generate1.Trade@51532e9f
handler3: name: h1 , price: 17.0;  instance: bhz.generate1.Trade@2c19f26f
handler3: name: h1 , price: 17.0;  instance: bhz.generate1.Trade@5b7b6c07
handler3: name: h1 , price: 17.0;  instance: bhz.generate1.Trade@3f16e7d
handler3: name: h1 , price: 17.0;  instance: bhz.generate1.Trade@5f8cd290
handler3: name: h1 , price: 17.0;  instance: bhz.generate1.Trade@16541cf2
handler3: name: h1 , price: 17.0;  instance: bhz.generate1.Trade@6c385f51
handler3: name: h1 , price: 17.0;  instance: bhz.generate1.Trade@68f9f658
handler3: name: h1 , price: 17.0;  instance: bhz.generate1.Trade@729c413f
handler3: name: h1 , price: 17.0;  instance: bhz.generate1.Trade@707fc9e1
总耗时:20049
												


																

最新文章
	

    
						
  1. android 学习第一天  了解事件机制，页面跳转等常用操作
		
    2. 
						
  2. [python IDE] 舒服的pycharm设置
		
    3. 
						
  3. homework-05
		
    4. 
						
  4. Codeforces Round #328 div2
		
    5. 
						
  5. js 一个关于图片onload加载的事
		
    6. 
						
  6. Codeforces_776E: The Holmes Children （数论 欧拉函数）
		
    7. 
						
  7. CSS属性速查表
		
    8. 
						
  8. Django认证系统
		
    9. 
						
  9. Balanced Ternary String CodeForces - 1102D （贪心+思维）
		
    10. 
						
  10. JS——取消事件冒泡，实现div的显示与隐藏 event.cancelBubble = true;
		
    11. 
						
  11. springboot + mybatis配置分页插件
		
    12. 
						
  12. opencvbase 实现opencv打开摄像头和初步处理等效果操作（附源码）
		
    13. 
						
  13. Ubuntu安装时怎样分区
		
    14. 
						
  14. $@和 $*-linux_Shell
		
    15. 
						
  15. Dubbo学习参考
		
    16. 
						
  16. MergeSort 归并排序（java）
		
    17. 
						
  17. Python中的相对文件路径的调用
		
    18. 
						
  18. 为eclipse配置javap命令
		
    19. 
						
  19. jquery页面初始化控件时间
		
    20. 
						
  20. 让ubuntu的ssh保持长时间连接
		
    21. 
		

	


热门文章
	

    
									
  1. MVC案例
			
    2. 
						
  2. zabbix通过IPMI模式监控服务器风扇转速和温度反映机房室温变化实例
			
    3. 
						
  3. PAT 1036 Boys vs Girls (25分) 比大小而已
			
    4. 
						
  4. 非阻塞赋值（Non-blocking Assignment）是个伪需求(2)
			
    5. 
						
  5. java方法句柄-----1.方法句柄类型、调用
			
    6. 
						
  6. Java实现 LeetCode 99 恢复二叉搜索树
			
    7. 
						
  7. 第八届蓝桥杯JavaB组国(决)赛真题
			
    8. 
						
  8. Java实现简易计算器
			
    9. 
						
  9. Java实现LeetCode17. 电话号码的字母组合
			
    10. 
						
  10. Java实现 洛谷 P1328 生活大爆炸版石头剪刀布
			
    11.