必威体育Betway必威体育官网
当前位置:首页 > IT技术

无锁并行计算框架Disruptor

时间:2019-07-07 12:42:09来源:IT技术作者:seo实验室小编阅读:62次「手机版」
 

disruptor

github地址

https://github.com/LMAX-Exchange/disruptor

Disruptor是一个javablockingqueue,它的目的是在相同处理器的不同线程之间move数据message or event)。相对于一般的queue,Disruptor具有一些重要的特性:为事件预分配内存、可选的lock-free、根据消费依赖路径多路发送事件给消费者

Disruptor说明文档:https://github.com/LMAX-Exchange/disruptor/wiki/Introduction

LMAX是一种新型零售金融交易平台,它能够以很低的延迟产生大量交易。这个系统是建立在JVM平台上,其核心是一个业务逻辑处理器,它能够在一个线程里每秒处理600万订单。业务逻辑处理器完全是运行在内存中,使用事件源驱动的方式。业务逻辑驱动器的核心是Disruptor。Disruptor主动将数据发送到数据消费端,600万指的是分发给消费端的订单数。

Disruptor是一个开源的并发框架,并获得2011年的程序框架创新奖,能够在无锁的情况下实现网络Queue并发操作。Disruptor是一个高性能的异步处理框架,或者认为是最快的消息框架(轻量的JMS),也可以是一个观察者模式的实现,或者事件监听模式的实现。

Disruptor相关术语:

RingBuffer:它是Disruptor最主要的组件,从3.0开始RingBuffer仅仅负责存储和更新在Disruptor中流通的数据,对一些特殊的使用场景能够被用户(使用其他数据结构)完全替代。

sequence:表示一个特殊组件处理的序号,每个消费者(EventProcessor)都维持着一个Sequence,大部分的并发代码依赖这些Sequence值运转,因此Sequence支持多种当前为AtomicLong类的特性。

Sequencer:这是Disruptor真正的核心,实现了这个接口的两种生产者(单生产者和多生产者)均实现了所有的并发算法,为了在生产者和消费者之间进行准确快速的数据传递。

SequenceBarrier:由Sequencer生成,并且包含了已经发布的Sequence的引用,它包含了决定是否有供消费者来消费的Event的逻辑。

WaitStrategy:决定一个消费者将如何等待生产者将Event置入Disruptor。

Event:从生产者到消费者过程中所处理的数据单元,它完全由用户自己定义。

eventhandler:由用户实现并且代表了Disruptor中的一个消费者的接口。

producer:由用户实现,它调用RingBuffer来插入事件,在Disruptor中没有其相应的实现代码,由用户实现。

WorkProcessor:确保每个sequence只被一个processor消费,在同一个WorkPool中处理多个WorkerProcessor不会消费同样的sequence。

WorkerPool:一个WorkProcessor池,其中WorkProcessor将消费sequence,所以任务可以在实现了WorkHandler接口的worker之间移交。

RingBuffer是一个首尾相接的环,可以把它作为在不同上下文(线程)之间传递数据的buffer。RingBuffer中拥有一个序号Sequence,它指向下一个可用元素。随着Producer不停地填充数据到这个buffer(可能也会有读取),这个序号会一直增长,直到绕过这个环。要找到数组中当前序号指向的元素,可以通过mod操作实现,即sequence mod array length = array index 。RingBuffer槽的个数最好为2的n次方,这样有利于基于二进制的计算机进行计算。RingBuffer使用的是数组的顺序存储结构,所以比链表更快。

在Disruptor中,创建一个HelloWorld示例程序的步骤:

1.建立一个Event类

2.建立一个工厂Event类,用于创建Event类实例对象

3.建立一个监听事件类,用于消费处理Event类型的数据

4.main函数测试代码编写,实例化Disruptor实例,配置一系列参数。然后,对Disruptor实例绑定监听事件类,接收并处理数据

5.在Disruptor中,真正存储数据的核心叫做RingBuffer,通过Disruptor实例获取RingBuffer,生产者Producer将数据生产出来,将数据加入到RingBuffer的槽中

LongEvent.java

public class LongEvent { 
    private long value;
    public long getValue() { 
        return value; 
    } 
 
    public void setValue(long value) { 
        this.value = value; 
    } 
} 

LongEventFactory.java

// 需要让disruptor为我们创建事件,我们同时还声明了一个EventFactory来实例化Event对象。
public class LongEventFactory implements EventFactory { 

    @Override 
    public Object newinstance() { 
        return new LongEvent(); 
    } 
} 
LongEventProducer.java
import java.nio.ByteBuffer;
import com.lmax.disruptor.RingBuffer;
/**
 * 发布事件最少需要两步:获取下一个事件槽并发布事件(发布事件的时候要使用try/finally保证事件一定会被发布)。
 * 如果我们使用RingBuffer.next()获取一个事件槽,那么一定要发布对应的事件。
 * 如果不能发布事件,那么就会引起Disruptor状态的混乱。
 * 尤其是在多个事件生产者的情况下会导致事件消费者失速,从而不得不重启应用才能会恢复。
 */
public class LongEventProducer {

	private final RingBuffer<LongEvent> ringBuffer;
	
	public LongEventProducer(RingBuffer<LongEvent> ringBuffer){
		this.ringBuffer = ringBuffer;
	}
	
	/**
	 * onData用来发布事件,每调用一次就发布一次事件
	 * 它的参数用事件传递给消费者
	 */
	public void onData(ByteBuffer bb){
		//1.可以把ringBuffer看做一个事件队列,那么next就是得到下面一个事件槽
		long sequence = ringBuffer.next();
		try {
			//2.用上面的索引取出一个空的事件用于填充(获取该序号对应的事件对象)
			LongEvent event = ringBuffer.get(sequence);
			//3.获取要通过事件传递的业务数据
			event.setValue(bb.getLong(0));
		} finally {
			//4.发布事件
			//注意,最后的 ringBuffer.publish 方法必须包含在 finally 中以确保必须得到调用;
            //某个请求的 sequence 未被提交,将会堵塞后续的发布操作或者其它的 producer。
			ringBuffer.publish(sequence);
		}
	}
}
LongEventHandler.java
import com.lmax.disruptor.EventHandler;

//我们还需要一个事件消费者,也就是一个事件处理器。这个事件处理器简单地把事件中存储的数据打印到终端:
public class LongEventHandler implements EventHandler<LongEvent>  {

	@Override
	public void onEvent(LongEvent longEvent, long l, boolean b) throws Exception {
		System.out.println(longEvent.getValue()); 		
	}

}
LongEventMain.java
import java.nio.ByteBuffer;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.YieldingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;

public class LongEventMain {

	public static void main(String[] args) throws Exception {
		//创建线程池
		ExecutorService  executor = Executors.newcachedThreadPool();
		//创建LongEvent工厂
		LongEventFactory factory = new LongEventFactory();
		//创建bufferSize ,也就是RingBuffer大小,必须是2的N次方
		int ringBufferSize = 1024 * 1024; // 

		/**
		//BlockingWaitStrategy 是最低效的策略,但其对cpu的消耗最小并且在各种不同部署环境中能提供更加一致的性能表现
		WaitStrategy BLOCKING_WAIT = new BlockingWaitStrategy();
		//SleepingWaitStrategy 的性能表现跟BlockingWaitStrategy差不多,对CPU的消耗也类似,但其对生产者线程的影响最小,适合用于异步日志类似的场景
		WaitStrategy SLEEPING_WAIT = new SleepingWaitStrategy();
		//YieldingWaitStrategy 的性能是最好的,适合用于低延迟的系统。在要求极高性能且事件处理线数小于CPU逻辑核心数的场景中,推荐使用此策略;例如,CPU开启超线程的特性
		WaitStrategy YIELDING_WAIT = new YieldingWaitStrategy();
		*/
		
		//创建disruptor,LongEvent为指定的消费数据类型,即每个槽中存放的数据类型
		Disruptor<LongEvent> disruptor = 
				new Disruptor<LongEvent>(factory, ringBufferSize, executor, ProducerType.SINGLE, new YieldingWaitStrategy());
		// 连接消费事件方法
		disruptor.handleEventsWith(new LongEventHandler());
		
		// 启动
		disruptor.start();
		
		//获取具体存放数据的ringbuffer(环形结构)
		RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();
		
		LongEventProducer producer = new LongEventProducer(ringBuffer); 
		//LongEventProducerWithTranslator producer = new LongEventProducerWithTranslator(ringBuffer);
		ByteBuffer byteBuffer = ByteBuffer.allocate(8);
		for(long i = 0; i<10; i++){
			byteBuffer.putLong(0, i);
			producer.onData(byteBuffer);
		}

		disruptor.shutdown();//关闭 disruptor,方法会堵塞,直至所有的事件都得到处理;
		executor.shutdown();//关闭使用的线程池;如果需要的话,必须手动关闭, disruptor在 shutdown时不会自动关闭线程池;		
				
	}
}
Eclipseconsole输出

在LongEventMain.java中,将代码

LongEventProducer producer = new LongEventProducer(ringBuffer);

修改为

LongEventProducerWithTranslator producer = new LongEventProducerWithTranslator(ringBuffer);

同时,将LongEventProducer.java修改为LongEventProducerWithTranslator.java

import java.nio.ByteBuffer;
import com.lmax.disruptor.EventTranslatorOneArg;
import com.lmax.disruptor.RingBuffer;

/**
 * Disruptor 3.0提供了lambda式的API。这样可以把一些复杂的操作放在Ring Buffer,
 * 所以在Disruptor3.0以后的版本最好使用Event Publisher或者Event Translator来发布事件
 */
public class LongEventProducerWithTranslator {

	//一个translator可以看做一个事件初始化器,publicEvent方法会调用它
	//填充Event
	private static final EventTranslatorOneArg<LongEvent, ByteBuffer> TRANSLATOR = 
			new EventTranslatorOneArg<LongEvent, ByteBuffer>() {
				@Override
				public void translateTo(LongEvent event, long sequeue, ByteBuffer buffer) {
					event.setValue(buffer.getLong(0));
				}
			};
	
	private final RingBuffer<LongEvent> ringBuffer;
	
	public LongEventProducerWithTranslator(RingBuffer<LongEvent> ringBuffer) {
		this.ringBuffer = ringBuffer;
	}
	
	public void onData(ByteBuffer buffer){
		ringBuffer.publishEvent(TRANSLATOR, buffer);
	}
}
在Eclipse的console输出如下:

直接使用RingBuffer(使用消息处理器)

Main1.java

import java.util.concurrent.Callable;
import java.util.concurrent.executionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import com.lmax.disruptor.BATchEventProcessor;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.EventProcessor;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.SequenceBarrier;
import com.lmax.disruptor.YieldingWaitStrategy;

public class Main1 {  
   
	public static void main(String[] args) throws Exception {  
        int BUFFER_SIZE=1024;  
        int THREAD_NUMBERS=4;  
        /* 
         * createSingleProducer创建一个单生产者的RingBuffer, 
         * 第一个参数叫EventFactory,从名字上理解就是"事件工厂",其实它的职责就是产生数据填充RingBuffer的区块。 
         * 第二个参数是RingBuffer的大小,它必须是2的指数倍 目的是为了将求模运算转为&运算提高效率 
         * 第三个参数是RingBuffer的生产都在没有可用区块的时候(可能是消费者(或者说是事件处理器) 太慢了)的等待策略 
         */  
        final RingBuffer<Trade> ringBuffer = RingBuffer.createSingleProducer(new EventFactory<Trade>() {  
            @Override  
            public Trade newInstance() {  
                return new Trade();  
            }  
        }, BUFFER_SIZE, new YieldingWaitStrategy());  
        
        //创建线程池  
        ExecutorService executors = Executors.newFixedThreadPool(THREAD_NUMBERS);  
        
        //创建SequenceBarrier  
        SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();  
          
        //创建消息处理器  
        BatchEventProcessor<Trade> transProcessor = new BatchEventProcessor<Trade>(  
                ringBuffer, sequenceBarrier, new TradeHandler());  
          
        //这一步的目的就是把消费者的位置信息引用注入到生产者    如果只有一个消费者的情况可以省略 
        ringBuffer.addGatingSequences(transProcessor.getSequence());  
          
        //把消息处理器提交到线程池  
        executors.submit(transProcessor);  
        
        System.out.println("主线程名"+Thread.currentThread().getName());
        //如果存在多个消费者 那重复执行上面3行代码 把TradeHandler换成其它消费者类  
          
        Future<?> future= executors.submit(new Callable<Void>() { //提交给线程池 
            @Override  
            public Void call() throws Exception {  
                long seq;  
                System.out.println("执行call()方法的线程名"+Thread.currentThread().getName());
                for(int i=0;i<10;i++){  
                    seq = ringBuffer.next();//占个坑 --ringBuffer一个可用区块  
                    ringBuffer.get(seq).setPrice(Math.random()*9999);//给这个区块放入 数据 
                    ringBuffer.publish(seq);//发布这个区块的数据使handler(consumer)可见  
                }  
                return null;  
            }  
        }); 
        
        future.get();//等待生产者结束  
        Thread.sleep(1000);//等上1秒,等消费都处理完成  
        transProcessor.halt();//通知事件(或者说消息)处理器 可以结束了(并不是马上结束!!!)  
        executors.shutdown();//终止线程  
    }  
}  

Trade.java

import java.util.concurrent.atomic.Atomicinteger;

public class Trade {  
	
	private String id;//ID  
	private String name;
	private double price;//金额  
	private AtomicInteger count = new AtomicInteger(0);
	
	public String getId() {
		return id;
	}
	public void setId(String id) {
		this.id = id;
	}
	public String getName() {
		return name;
	}
	public void setName(String name) {
		this.name = name;
	}
	public double getPrice() {
		return price;
	}
	public void setPrice(double price) {
		this.price = price;
	}
	public AtomicInteger getCount() {
		return count;
	}
	public void setCount(AtomicInteger count) {
		this.count = count;
	} 
	  
	  
}  

TradeHandler.java

import java.util.UUID;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.WorkHandler;

public class TradeHandler implements EventHandler<Trade> {  
	  
    @Override  
    public void onEvent(Trade event, long sequence, boolean endOfBatch) throws Exception {  
    	System.out.println("执行TradeHandler中onEvent(Trade event)方法的线程名"+Thread.currentThread().getName());
        //这里做具体的消费逻辑  
        event.setId(UUID.randomUUID().toString());//简单生成下ID  
        System.out.println(event.getId());    
    }  
}

在main主线程中创建单生产者的RingBuffer,使用了线程池来进行并发线程控制。涉及Future与Callable的使用,在main线程中创建子线程,执行call()方法,向RingBuffer中填充Trade类型的数据,并publish相应的sequence,使消费者TradeHandler对放到RingBuffer上的数据可见。执行代码future.get(),使主线程main阻塞等待子线程执行完毕。执行transProcessor.halt();等待消费者将消息消费完毕。最后,关闭线程池。

Eclipse中console控制台输出

直接使用RingBuffer(使用WorkHandler与WorkerPool)

Main2.java

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.ignoreExceptionHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.SequenceBarrier;
import com.lmax.disruptor.WorkHandler;
import com.lmax.disruptor.WorkerPool;

public class Main2 {  
    public static void main(String[] args) throws InterruptedException {  
        int BUFFER_SIZE=1024;  
        int THREAD_NUMBERS=4;  
        
        EventFactory<Trade> eventFactory = new EventFactory<Trade>() {  
            public Trade newInstance() {  
                return new Trade();  
            }  
        };  
        
        RingBuffer<Trade> ringBuffer = RingBuffer.createSingleProducer(eventFactory, BUFFER_SIZE);  
          
        SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();  
          
        ExecutorService executor = Executors.newFixedThreadPool(THREAD_NUMBERS);  
          
        WorkHandler<Trade> handler = new TradeHandler();//充当消费者  

        WorkerPool<Trade> workerPool = new WorkerPool<Trade>(ringBuffer, sequenceBarrier, new IgnoreExceptionHandler(), handler);  
          
        workerPool.start(executor);  
          
        //下面这个生产8个数据
        for(int i=0;i<8;i++){  
            long seq=ringBuffer.next();  
            ringBuffer.get(seq).setPrice(Math.random()*9999);  
            ringBuffer.publish(seq);  
        }  
          
        Thread.sleep(1000);  
        workerPool.halt();  
        executor.shutdown();  
    }  
}  

TradeHandler.java

import java.util.UUID;
import com.lmax.disruptor.WorkHandler;

public class TradeHandler implements  WorkHandler<Trade> {  
	  
    @Override  
    public void onEvent(Trade event) throws Exception {  
    	System.out.println("执行TradeHandler中onEvent(Trade event)方法的线程名"+Thread.currentThread().getName());
        //这里做具体的消费逻辑  
        event.setId(UUID.randomUUID().toString());//简单生成下ID  
        System.out.println(event.getId());  
    }  
}  
Eclipse的console输出:

使用Disruptor实现菱形、六边形操作

Main.java

import java.util.concurrent.countdownlatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
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));  
    }  
} 

TradePublisher.java

import java.util.Random;
import java.util.concurrent.CountDownLatch;
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=1;//模拟百万次交易的发生  
  
    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;  
    }  
	
}  

Handler1.java

import java.util.UUID;
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);
    }  
}  

Handler2.java

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);
    }  
      
}  

Handler3.java

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());
    }  
}

Eclipse的console输出如下:

Disruptor实现多生产者多消费者

order为Disruptor中的RingBuffer环中每一个槽存放数据的类型。Producer用于产生事件并放置到RingBuffer事件槽中。Consumer是用于消费RingBuffer中事件的类。

Order.java

public class Order {  
	
	private String id;//ID  
	private String name;
	private double price;//金额  
	
	public String getId() {
		return id;
	}
	public void setId(String id) {
		this.id = id;
	}
	public String getName() {
		return name;
	}
	public void setName(String name) {
		this.name = name;
	}
	public double getPrice() {
		return price;
	}
	public void setPrice(double price) {
		this.price = price;
	}
	  
}  

Producer.java

import java.nio.ByteBuffer;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicInteger;
import com.lmax.disruptor.EventTranslatorOneArg;
import com.lmax.disruptor.RingBuffer;

public class Producer {

	private final RingBuffer<Order> ringBuffer;
	
	public Producer(RingBuffer<Order> ringBuffer){
		this.ringBuffer = ringBuffer;
	}
	
	/**
	 * onData用来发布事件,每调用一次就发布一次事件
	 * 它的参数会用过事件传递给消费者
	 */
	public void onData(String data){
		//可以把ringBuffer看做一个事件队列,那么next就是得到下面一个事件槽
		long sequence = ringBuffer.next();
		try {
			//用上面的索引取出一个空的事件用于填充(获取该序号对应的事件对象)
			Order order = ringBuffer.get(sequence);
			//获取要通过事件传递的业务数据
			order.setId(data);
		} finally {
			//发布事件
			//注意,最后的 ringBuffer.publish 方法必须包含在 finally 中以确保必须得到调用;如果某个请求的 sequence 未被提交,将会堵塞后续的发布操作或者其它的 producer。
			ringBuffer.publish(sequence);
		}
	}
	
	
}

Consumer.java

import java.util.concurrent.atomic.AtomicInteger;
import com.lmax.disruptor.WorkHandler;

public class Consumer implements WorkHandler<Order>{
	
	private String consumerId;
	
	private static AtomicInteger count = new AtomicInteger(0);
	
	public Consumer(String consumerId){
		this.consumerId = consumerId;
	}

	@Override
	public void onEvent(Order order) throws Exception {
		System.out.println("当前消费者: " + this.consumerId + ",消费信息:" + order.getId());
		count.incrementAndGet();
	}
	
	public int getCount(){
		return count.get();
	}

}
Main.java
import java.nio.ByteBuffer;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.ExceptionHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.SequenceBarrier;
import com.lmax.disruptor.WorkHandler;
import com.lmax.disruptor.WorkerPool;
import com.lmax.disruptor.YieldingWaitStrategy;
import com.lmax.disruptor.dsl.ProducerType;

public class Main {
	//提供Integer的原子操作,在高并发场景中,多线程操作能够保证线程安全
	private static AtomicInteger count = new AtomicInteger(0);
	
	public static void main(String[] args) throws Exception {

		//创建ringBuffer
		RingBuffer<Order> ringBuffer = 
				RingBuffer.create(ProducerType.MULTI, 
						new EventFactory<Order>() {  
				            @Override  
				            public Order newInstance() {  
				                return new Order();  
				            }  
				        }, 
				        1024 * 1024, 
						new YieldingWaitStrategy());
		
		SequenceBarrier barriers = ringBuffer.newBarrier();//生产者消费者协调操作
		
		//消费者数组
		Consumer[] consumers = new Consumer[3];
		for(int i = 0; i < consumers.length; i++){
			consumers[i] = new Consumer("c" + i);
		}
		
		WorkerPool<Order> workerPool = 
				new WorkerPool<Order>(ringBuffer, 
						barriers, 
						new IntEventExceptionHandler(),
						consumers);
		
		//这一步的目的就是把消费者的位置信息引用注入到生产者,协调生产与消费的进度
        ringBuffer.addGatingSequences(workerPool.getWorkerSequences());  
        workerPool.start(Executors.newFixedThreadPool(runtime.getRuntime().availableProcessors()));  
        
        final CountDownLatch latch = new CountDownLatch(1);
        for (int i = 0; i < 3; i++) {  
        	final Producer p = new Producer(ringBuffer);
        	new Thread(new Runnable() {//for循环创建3个线程,每个线程内部的for循环产生2个事件并publish
				@Override
				public void run() {
					try {
						System.out.println(Thread.currentThread().getName()+"ready!");
						latch.await();//子线程阻塞
					} catch (InterruptedException e) {
						e.printstacktrace();
					}
					for(int j = 0; j < 2; j ++){
						p.onData(Thread.currentThread().getName()+"产生的事件(数据)"+String.valueOf(count.addAndGet(1)));
					}
				}
			}).start();
        } 
        Thread.sleep(2000);
        System.out.println("---------------开始生产-----------------");
        latch.countDown();//解除对上面2个生产者子线程的阻塞,使其开始产生和publish事件
        Thread.sleep(5000);
        System.out.println("总数:" + consumers[0].getCount() );
	}
	
	static class IntEventExceptionHandler implements ExceptionHandler {  
	    public void handleEventException(throwable ex, long sequence, Object event) {}  
	    public void handleOnStartException(Throwable ex) {}  
	    public void handleOnShutdownException(Throwable ex) {}  
	} 
}

Eclipse的console输出

相关阅读

eclipse 搭建ssm框架

由于工作需要,我在未来的一段时间里将要使用eclipse开发工具进行java后端的开发。我个人使用过eclipse和idea都有一段时间,个人觉得

NELL框架

文章目录Never-Ending LearningAbstractIntroductionRelated WorkNerer-Ending-Learning举个例子,永动学习机NELL的学习问题实证评

如何做内容运营?先打好框架再说!

作为一个内容运营,我们要明确自身的内容定位、用户需求、产品调性等等因素,来确定自己的内容生产和流通机制,同时,不断监测运营数据变

管理后台界面基本框架设计

电商类项目分为前端系统与后台管理系统,前端系统面向终端用户使用,一般设计十分炫动,色彩对比比较鲜明,容易引起消费者注意力,需要由美

RPC框架pigeon源码分析

Pigeon是一个分布式服务通信框架(RPC),是美团点评最基础的底层框架之一。已开源,链接:https://github.com/dianping/pigeon从接下来三

分享到:

栏目导航

推荐阅读

热门阅读