您当前的位置： 首页 > 

Disruptor详解

发布时间：2020-10-11 20:21:37 ，浏览量：0

LMAX是一种新型零售金融交易平台，它能够以很低的延迟产生大量交易。这个系统是建立在JVM平台上，其核心是一个业务逻辑处理器，它能够在一个线程里每秒处理6百万订单。业务逻辑处理器完全是运行在内存中，使用事件源驱动方式。业务逻辑处理器的核心是Disruptor。 Disruptor它是一个开源的并发框架，并获得2011 Duke’s 程序框架创新奖，能够在无锁的情况下实现网络的Queue并发操作。 Disruptor是一个高性能的异步处理框架，或者可以认为是最快的消息框架（轻量的JMS），也可以认为是一个观察者模式的实现，或者事件监听模式的实现。 在Disruptor中，我们想实现hello world 需要如下几步骤： 第一：建立一个Event类 第二：建立一个工厂Event类，用于创建Event类实例对象 第三：需要有一个监听事件类，用于处理数据（Event类） 第四：我们需要进行测试代码编写。实例化Disruptor实例，配置一系列参数。然后我们对Disruptor实例绑定监听事件类，接受并处理数据。 第五：在Disruptor中，真正存储数据的核心叫做RingBuffer，我们通过Disruptor实例拿到它，然后把数据生产出来，把数据加入到RingBuffer的实例对象中即可。 Event类：数据封装类 

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

工厂Event类：实现EventFactory<>接口的实现类 

public class LongEventFactory implements EventFactory<LongEvent>{ @Override
	public LongEvent newInstance() { return new LongEvent(); } }

EventHandler类：数据处理类实现EventHandler<>接口 

/**
 * 消费者，事件监听
 * @author Administrator
 *
 */
public class LongEventHandler implements EventHandler<LongEvent>{ @Override
	public void onEvent(LongEvent longEvent, long l, boolean b) throws Exception { //消费，数据处理
		System.out.println(longEvent.getValue()); } }

数据生产类： 

public class LongEventProducer { private final RingBuffer<LongEvent> ringBuffer; public LongEventProducer(RingBuffer<LongEvent> ringBuffer) { this.ringBuffer=ringBuffer; } public void onData(ByteBuffer bb) { //可以把ringBuffer看做一个事件队列，那么next就是得到下面一个事件槽
		long sequence=ringBuffer.next(); try { //用上面的索引取出一个空的事件用于填充 
			LongEvent l=ringBuffer.get(sequence); l.setValue(bb.getLong( 0 )); }catch (Exception e) { }finally { ringBuffer.publish(sequence); } } }

测试类： 

public class LongEventTest { public static void main(String[] args) { ExecutorService executor=Executors.newCachedThreadPool(); LongEventFactory eventFactory=new LongEventFactory(); //必须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> dis=new Disruptor<>(eventFactory, ringBufferSize, executor, ProducerType.SINGLE, new YieldingWaitStrategy()); dis.handleEventsWith(new LongEventHandler()); dis.start(); RingBuffer<LongEvent> ringBuffer=dis.getRingBuffer(); LongEventProducer producer=new LongEventProducer(ringBuffer); //LongEventProducerWithTranslator producer = new LongEventProducerWithTranslator(ringBuffer); ByteBuffer bb=ByteBuffer.allocate( 8 ); for (int i = 0 ; i < 100 ; i++) { bb.putLong( 0
,i); producer.onData(bb); } dis.shutdown(); executor.shutdown(); } } EventProducerWithTranslator实现方式：

public class LongEventProducerWithTranslator { //一个translator可以看做一个事件初始化器，publicEvent方法会调用它
	//填充Event
	private static final EventTranslatorOneArg<LongEvent, ByteBuffer> TRANSLATOR= new EventTranslatorOneArg<LongEvent, ByteBuffer>() { @Override
		public void translateTo(LongEvent event, long sequence, 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); } }
Disruptor术语说明

RingBuffer: 被看作Disruptor最主要的组件，然而从3.0开始RingBuffer仅仅负责存储和更新在Disruptor中流通的数据。对一些特殊的使用场景能够被用户(使用其他数据结构)完全替代。 Sequence: Disruptor使用Sequence来表示一个特殊组件处理的序号。和Disruptor一样，每个消费者(EventProcessor)都维持着一个Sequence。大部分的并发代码依赖这些Sequence值的运转，因此Sequence支持多种当前为AtomicLong类的特性。 Sequencer: 这是Disruptor真正的核心。实现了这个接口的两种生产者（单生产者和多生产者）均实现了所有的并发算法，为了在生产者和消费者之间进行准确快速的数据传递。 SequenceBarrier: 由Sequencer生成，并且包含了已经发布的Sequence的引用，这些的Sequence源于Sequencer和一些独立的消费者的Sequence。它包含了决定是否有供消费者来消费的Event的逻辑。 WaitStrategy：决定一个消费者将如何等待生产者将Event置入Disruptor。 Event：从生产者到消费者过程中所处理的数据单元。Disruptor中没有代码表示Event，因为它完全是由用户定义的。 EventProcessor：主要事件循环，处理Disruptor中的Event，并且拥有消费者的Sequence。它有一个实现类是BatchEventProcessor，包含了event loop有效的实现，并且将回调到一个EventHandler接口的实现对象。 EventHandler：由用户实现并且代表了Disruptor中的一个消费者的接口。 Producer：由用户实现，它调用RingBuffer来插入事件(Event)，在Disruptor中没有相应的实现代码，由用户实现。 WorkProcessor：确保每个sequence只被一个processor消费，在同一个WorkPool中的处理多个WorkProcessor不会消费同样的sequence。 WorkerPool：一个WorkProcessor池，其中WorkProcessor将消费Sequence，所以任务可以在实现WorkHandler接口的worker吃间移交 LifecycleAware：当BatchEventProcessor启动和停止时，于实现这个接口用于接收通知。

EventProcessor使用： handler消费类： 

public class TradeHandler implements EventHandler<Trade>,WorkHandler<Trade>{ @Override
	public void onEvent(Trade event) throws Exception { //生成订单id
		event.setId(UUID.randomUUID().toString()); System.out.println(event); } @Override
	public void onEvent(Trade event, long sequence, boolean endOfBatch) throws Exception { this.onEvent(event); } }

Trade数据封装类： 

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

EventProcessorMain测试类： 

public static void main(String[] args) throws InterruptedException, ExecutionException { 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); //如果存在多个消费者，那么重复执行上面三行代码，把TradeHandler换成其他消费者类
		Future<?> future=executors.submit(new Callable<Trade>() { @Override
			public Trade call() throws Exception { long seq; 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 );//等待一秒，等消费者处理完成
		transProcessor.halt();//通知事件（或者说消息）处理器，可以结束了（并不是马上结束）
		executors.shutdown();//终止线程 }

WorkProcessor使用： WorkProcessorMain测试类： 

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.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 WorkProcessorMain { public static void main(String[] args) throws InterruptedException, ExecutionException { int BUFFER_SIZE = 1024 ; int THREAD_NUMBERS = 4 ; EventFactory<Trade> eventFactory = new EventFactory<Trade>() { @Override
			public Trade newInstance() { return new Trade(); } }; final RingBuffer<Trade> ringBuffer = RingBuffer.createSingleProducer(eventFactory, BUFFER_SIZE); SequenceBarrier sequenceBarrier = ringBuffer.newBarrier(); ExecutorService executors = Executors.newFixedThreadPool(THREAD_NUMBERS); WorkHandler<Trade> handler = new TradeHandler(); WorkerPool<Trade> workerPool = new WorkerPool<>(ringBuffer, sequenceBarrier, new IgnoreExceptionHandler(),
				handler); workerPool.start(executors); // 如果存在多个消费者，那么重复执行上面三行代码，把TradeHandler换成其他消费者类
		Future<?> future = executors.submit(new Callable<Trade>() { @Override
			public Trade call() throws Exception { long seq; 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 );// 等待一秒，等消费者处理完成
		workerPool.halt();// 通知事件（或者说消息）处理器，可以结束了（并不是马上结束）
		executors.shutdown();// 终止线程 } }
并行计算 - 多边形高端操作 菱形操作

Disruptor可实现串并行同时编码。

在复杂场景下使用RingBuffer（希望P1生产的数据给C1、C2并行执行，最后C1、C2执行结束后C3执行）

  • C1和C2并行执行。
六边形操作

C1h和C2并行执行，C4和C5并行执行，并行执行完后执行C3 示例： C1: 

import com.lmax.disruptor.EventHandler; import com.lmax.disruptor.WorkHandler; import com.moudle.disruptorDemo.generate1.Trade; public class Handler1 implements EventHandler<Trade>,WorkHandler<Trade>{ @Override
	public void onEvent(Trade trade) throws Exception { System.out.println("handler1 set name:"); trade.setName("h1"); Thread.sleep( 1000 ); } @Override
	public void onEvent(Trade arg0, long arg1, boolean arg2) throws Exception { this.onEvent(arg0); } }

C2 

import com.lmax.disruptor.EventHandler; import com.lmax.disruptor.WorkHandler; import com.moudle.disruptorDemo.generate1.Trade; public class Handler2 implements EventHandler<Trade>,WorkHandler<Trade>{ @Override
	public void onEvent(Trade trade) throws Exception { System.out.println("handler2 set price:"); trade.setPrice( 17 ); Thread.sleep( 1000 ); } @Override
	public void onEvent(Trade arg0, long arg1, boolean arg2) throws Exception { this.onEvent(arg0); } }

C3 

import com.lmax.disruptor.EventHandler; import com.lmax.disruptor.WorkHandler; import com.moudle.disruptorDemo.generate1.Trade; public class Handler3 implements EventHandler<Trade>,WorkHandler<Trade>{ @Override
	public void onEvent(Trade event) throws Exception { System.out.println("handler3: name: " + event.getName() + " , price: " + event.getPrice() + ";  instance: " + event.toString()); } @Override
	public void onEvent(Trade arg0, long arg1, boolean arg2) throws Exception { this.onEvent(arg0); } }

C4 

import com.lmax.disruptor.EventHandler; import com.lmax.disruptor.WorkHandler; import com.moudle.disruptorDemo.generate1.Trade; public class Handler4 implements EventHandler<Trade>,WorkHandler<Trade>{ @Override
	public void onEvent(Trade trade) throws Exception { System.out.println("handler4 set addName:"); trade.setName(trade.getName()+"h4"); } @Override
	public void onEvent(Trade arg0, long arg1, boolean arg2) throws Exception { this.onEvent(arg0); } }

C5 

import com.lmax.disruptor.EventHandler; import com.lmax.disruptor.WorkHandler; import com.moudle.disruptorDemo.generate1.Trade; public class Handler5 implements EventHandler<Trade>,WorkHandler<Trade>{ @Override
	public void onEvent(Trade trade) throws Exception { System.out.println("handler5 set add price:"); trade.setPrice(trade.getPrice()+
3 ); } @Override
	public void onEvent(Trade arg0, long arg1, boolean arg2) throws Exception { this.onEvent(arg0); } }

P1(生产者) 

import java.util.Random; import java.util.concurrent.CountDownLatch; import com.lmax.disruptor.EventTranslator; import com.lmax.disruptor.dsl.Disruptor; import com.moudle.disruptorDemo.generate1.Trade; public class TradePublisher implements Runnable{ Disruptor<Trade> disruptor; private CountDownLatch latch; private static int count = 1 ;//模拟百万次交易的发生
	
	public TradePublisher(Disruptor<Trade> disruptor,CountDownLatch latch){ this.disruptor=disruptor; this.latch=latch; } @Override
	public void run() { TradeEventTranslator translator=new TradeEventTranslator(); for(int i= 0 ;i<count;i++){ disruptor.publishEvent(translator); } latch.countDown(); } } class TradeEventTranslator implements EventTranslator<Trade>{ private Random random=new Random(); @Override
	public void translateTo(Trade trade, long arg1) { this.generateTrade(trade); } private Trade generateTrade(Trade trade){ trade.setPrice(random.nextDouble()*
9999 ); return trade; } }

Main: 

package com.moudle.disruptorDemo.generate2; 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; import com.moudle.disruptorDemo.generate1.Trade; 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<>(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()); //輸出結果：
//		handler1 set name:
//		handler2 set price:
//		handler3: name: h1 , price: 17.0; instance: com.moudle.disruptorDemo.generate1.Trade@220a5c4d
		
		/*//六边形操作
		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); //输出结果：
//        handler1 set name:
//        handler2 set price:
//        handler4 set addName:
//        handler5 set add price:
//        handler3: name: h1h4 , price: 20.0; instance: com.moudle.disruptorDemo.generate1.Trade@5e6d6957
        */
	/*	//顺序执行
		disruptor.handleEventsWith(new Handler1()).
    	handleEventsWith(new Handler2()).
    	handleEventsWith(new Handler3()); //输出结果：
//			handler1 set name:
//			handler2 set price:
//			handler3: name: h1 , price: 17.0; instance: com.moudle.disruptorDemo.generate1.Trade@331d6441
		*/
		disruptor.start();//启动
		CountDownLatch latch=new CountDownLatch( 1 ); //生产者准备
		executor.submit(new TradePublisher(disruptor, latch)); latch.await();//等待生产完成
		disruptor.shutdown(); executor.shutdown(); } }

多生产者多消费者的使用： Order订单类: 

package com.moudle.disruptorDemo.multi; 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生产者: 

package com.moudle.disruptorDemo.multi; 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); } catch (Exception e) { }finally{ //发布事件
			//注意，最后的 ringBuffer.publish 方法必须包含在 finally 中以确保必须得到调用；如果某个请求的 sequence 未被提交，将会堵塞后续的发布操作或者其它的 producer。
			ringBuffer.publish(sequence); } } }

Consumer消费者： 

package com.moudle.disruptorDemo.multi; 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); count.incrementAndGet(); } public int getCount(){ return count.get(); } }

Main测试类： 

package com.moudle.disruptorDemo.multi; import java.util.UUID; import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executor; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import com.lmax.disruptor.EventFactory; import com.lmax.disruptor.ExceptionHandler; import com.lmax.disruptor.RingBuffer; import com.lmax.disruptor.SequenceBarrier; import com.lmax.disruptor.WorkerPool; import com.lmax.disruptor.YieldingWaitStrategy; import com.lmax.disruptor.dsl.ExceptionHandlerWrapper; import com.lmax.disruptor.dsl.ProducerType; public class Main { public static void main(String[] args) throws Exception{ //		RingBuffer<Order> ringBuffer=RingBuffer.create( //				ProducerType.MULTI, new EventFactory<Order>() { //					@Override
//					public Order newInstance() { // return new Order(); // } // }, 1024*1024, new YieldingWaitStrategy()); //创建ringBuffer
		RingBuffer<Order> ringBuffer=RingBuffer.createMultiProducer(new EventFactory<Order>() { @Override
			public Order newInstance() { return new Order(); } }, 
1024
*
1024
, new YieldingWaitStrategy()); //创建SequenceBarrier
		SequenceBarrier barriers=ringBuffer.newBarrier(); //创建3个消费者实例
		Consumer[] consumers=new Consumer[ 3 ]; for (int i = 0 ; i < consumers.length; i++) { consumers[i]=new Consumer("c"+i); } WorkerPool<Order> workerPool=new WorkerPool<>( ringBuffer, barriers, new IntEventExceptionHandler(), consumers); //这一步的目的是把消费者的位置信息引用注入到生产者 如果只有一个消费者的情况可以省略。
		//workerPool.getWorkerSequences()获取Sequence集合
		ringBuffer.addGatingSequences(workerPool.getWorkerSequences()); //创建线程池
		ExecutorService executorService=Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors()); workerPool.start(executorService); final CountDownLatch latch=new CountDownLatch( 1 ); for (int i = 0 ; i < 100 ; i++) { final Producer producer=new Producer(ringBuffer); new Thread(new Runnable() { @Override
				public void run() { try { //等待生产者100个线程启动
						latch.await(); for (int j = 0 ; j < 100 ; j++) { //生产数据 
							producer.onData(UUID.randomUUID().toString()); } } catch (InterruptedException e) { e.printStackTrace(); } } }).start(); } //等待两秒，等生产者的100个线程启动
		Thread.sleep( 2000 ); System.out.println("---------------开始生产-----------------"); latch.countDown(); Thread.sleep( 5000 ); System.out.println("总数:"+consumers[ 0 ].getCount()); executorService.shutdown(); } static class IntEventExceptionHandler implements ExceptionHandler<Order>{ @Override
		public void handleEventException(Throwable arg0, long arg1, Order arg2) { } @Override
		public void handleOnShutdownException(Throwable arg0) { } @Override
		public void handleOnStartException(Throwable arg0) { } } }

参考

  • https://programtip.com/zh/art-111663
关注
打赏
1688896170
查看更多评论

暂无认证

  • 0浏览

    0关注

    115984博文

    0收益

  • 0浏览

    0点赞

    0打赏

    0留言

私信
关注
热门博文

[ 申请 ]友情链接：

復刻手錶 快连vpn 快连 快连vpn 搜外友链 笔趣阁 爱思助手 ClashX教程 绘画宝宝 配音宝宝
立即登录/注册

微信扫码登录

0.0495s