文档章节

并发框架Disruptor几个Demo

chaun
 chaun
发布于 2016/11/02 17:20
字数 1462
阅读 29
收藏 3

经过2个月的疯狂加班后终于迎来了五一节前的几天清闲日子,在这闲得蛋疼的日子里,本屌丝无意中从ifeve上看到“Disruptor”并发框架,它号称"能够在一个线程里每秒处理6百万订单" 当时就被它吸引住了,不管是不是在吹流逼吧,先研究研究在说....

 

扫盲:

要想了解Disruptor框架必需多花点时间研究下它的工作原理,不然代码是没法撸的!!!

关于Disruptor的详细资料及原理请细看!!!    http://ifeve.com/disruptor/

 

Disruptor版本: 3.2.1

 

名词解释

消费者==事件处理器

 

一、现在你必须明白以下问题:

1、你必须明白Ringbuffer是什么,它的数据结构是怎么样的,有什么约定,为什么高效,它的职责是什么。

2、ConsumerBarrier (ifeve网上的译文版本比较早,这个类在2.0.0之后就一直被改名,3.2.1的版本中它是SequenceBarrier)它的职责是什么。

 

Disruptor框架在2.0版本之后不再采用生产者、消费者模型来编写API,而是使用事件模型,其实只是接口设计、类名和概念上的变化,内部原理其实还是一样的。

 

DEMO 一、使用原生API创建一个简单的生产者和消费者

 

Java代码 

//DEMO中使用的 消息全假定是一条交易
public class TradeTransaction {
	private String id;//交易ID
	private double price;//交易金额
	
	public TradeTransaction() {
	}
	public TradeTransaction(String id, double price) {
		super();
		this.id = id;
		this.price = price;
	}
	public String getId() {
		return id;
	}
	public void setId(String id) {
		this.id = id;
	}
	public double getPrice() {
		return price;
	}
	public void setPrice(double price) {
		this.price = price;
	}
}

public class TradeTransactionInDBHandler implements EventHandler<TradeTransaction>,WorkHandler<TradeTransaction> {

	@Override
	public void onEvent(TradeTransaction event, long sequence,
			boolean endOfBatch) throws Exception {
		this.onEvent(event);
	}

	@Override
	public void onEvent(TradeTransaction event) throws Exception {
		//这里做具体的消费逻辑
		event.setId(UUID.randomUUID().toString());//简单生成下ID
		System.out.println(event.getId());
	}
}



public class Demo1 {
	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<TradeTransaction> ringBuffer = RingBuffer.createSingleProducer(new EventFactory<TradeTransaction>() {
			@Override
			public TradeTransaction newInstance() {
				return new TradeTransaction();
			}
		}, BUFFER_SIZE,new YieldingWaitStrategy());
		//创建线程池
		ExecutorService executors = Executors.newFixedThreadPool(THREAD_NUMBERS);
		//创建SequenceBarrier
		SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();
		
		//创建消息处理器
		BatchEventProcessor<TradeTransaction> transProcessor = new BatchEventProcessor<TradeTransaction>(
				ringBuffer, sequenceBarrier, new TradeTransactionInDBHandler());
		
		//这一部的目的是让RingBuffer根据消费者的状态	如果只有一个消费者的情况可以省略
		ringBuffer.addGatingSequences(transProcessor.getSequence());
		
		//把消息处理器提交到线程池
		executors.submit(transProcessor);
		//如果存大多个消费者 那重复执行上面3行代码 把TradeTransactionInDBHandler换成其它消费者类
		
		Future<?> future=executors.submit(new Callable<Void>() {
			@Override
			public Void call() throws Exception {
				long seq;
				for(int i=0;i<1000;i++){
					seq=ringBuffer.next();//占个坑	--ringBuffer一个可用区块
					
					ringBuffer.get(seq).setPrice(Math.random()*9999);//给这个区块放入 数据  如果此处不理解,想想RingBuffer的结构图
					
					ringBuffer.publish(seq);//发布这个区块的数据使handler(consumer)可见
				}
				return null;
			}
		});
		future.get();//等待生产者结束
		Thread.sleep(1000);//等上1秒,等消费都处理完成
		transProcessor.halt();//通知事件(或者说消息)处理器 可以结束了(并不是马上结束!!!)
		executors.shutdown();//终止线程
	}
}

DEMO二、使用WorkerPool辅助创建消费者

Java代码

public class Demo2 {
	public static void main(String[] args) throws InterruptedException {
		int BUFFER_SIZE=1024;
		int THREAD_NUMBERS=4;
		EventFactory<TradeTransaction> eventFactory=new EventFactory<TradeTransaction>() {
			public TradeTransaction newInstance() {
				return new TradeTransaction();
			}
		};
		RingBuffer<TradeTransaction> ringBuffer=RingBuffer.createSingleProducer(eventFactory, BUFFER_SIZE);
		
		SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();
		
		ExecutorService executor = Executors.newFixedThreadPool(THREAD_NUMBERS);
		
		WorkHandler<TradeTransaction> workHandlers=new TradeTransactionInDBHandler();
		/*
		 * 这个类代码很简单的,亲自己看哈!~
		 */
		WorkerPool<TradeTransaction> workerPool=new WorkerPool<TradeTransaction>(ringBuffer, sequenceBarrier, new IgnoreExceptionHandler(), workHandlers);
		
		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();
	}
}

DEMO三、demo3写个流弊点的像下图这样。这次用Disruptor来完成整个构建工作.

从中图可以看出需求是介样子的:生产者生产数据经过C1,C2处理完成后再到C3。

假设如下场景:

1、交易网关收到交易(P1)把交易数据发到RingBuffer中,

2、负责处理增值业务的消费者C1和负责数据存储的消费者C2负责处理交易

3、负责发送JMS消息的消费者C3在C1和C2处理完成后再进行处理。

 

让代码说话:

Java代码

public class TradeTransactionJMSNotifyHandler implements EventHandler<TradeTransaction> {

	@Override
	public void onEvent(TradeTransaction event, long sequence,
			boolean endOfBatch) throws Exception {
		//do send jms message
	}
}



public class TradeTransactionPublisher implements Runnable{
	Disruptor<TradeTransaction> disruptor;
	private CountDownLatch latch;
	private static int LOOP=10000000;//模拟一千万次交易的发生

	public TradeTransactionPublisher(CountDownLatch latch,Disruptor<TradeTransaction> disruptor) {
		this.disruptor=disruptor;
		this.latch=latch;
	}

	@Override
	public void run() {
		TradeTransactionEventTranslator tradeTransloator=new TradeTransactionEventTranslator();
		for(int i=0;i<LOOP;i++){
			disruptor.publishEvent(tradeTransloator);
		}
		latch.countDown();
	}
	
}

class TradeTransactionEventTranslator implements EventTranslator<TradeTransaction>{
	private Random random=new Random();
	@Override
	public void translateTo(TradeTransaction event, long sequence) {
		this.generateTradeTransaction(event);
	}
	private TradeTransaction generateTradeTransaction(TradeTransaction trade){
		trade.setPrice(random.nextDouble()*9999);
		return trade;
	}
}


public class TradeTransactionVasConsumer implements EventHandler<TradeTransaction> {

	@Override
	public void onEvent(TradeTransaction event, long sequence,
			boolean endOfBatch) throws Exception {
		//do something....
	}
	
}



public class Demo3 {
	public static void main(String[] args) throws InterruptedException {
		long beginTime=System.currentTimeMillis();
		
		int bufferSize=1024;
		ExecutorService executor=Executors.newFixedThreadPool(4);
		//这个构造函数参数,相信你在了解上面2个demo之后就看下就明白了,不解释了~
		Disruptor<TradeTransaction> disruptor=new Disruptor<TradeTransaction>(new EventFactory<TradeTransaction>() {
			@Override
			public TradeTransaction newInstance() {
				return new TradeTransaction();
			}
		}, bufferSize, executor, ProducerType.SINGLE, new BusySpinWaitStrategy());
		
		//使用disruptor创建消费者组C1,C2
		EventHandlerGroup<TradeTransaction> handlerGroup=disruptor.handleEventsWith(new TradeTransactionVasConsumer(),new TradeTransactionInDBHandler());
		
		TradeTransactionJMSNotifyHandler jmsConsumer=new TradeTransactionJMSNotifyHandler();
		//声明在C1,C2完事之后执行JMS消息发送操作 也就是流程走到C3
		handlerGroup.then(jmsConsumer);
		
		
		disruptor.start();//启动
		CountDownLatch latch=new CountDownLatch(1);
		//生产者准备
		executor.submit(new TradeTransactionPublisher(latch, disruptor));
		latch.await();//等待生产者完事.
		disruptor.shutdown();
		executor.shutdown();
		
		System.out.println("总耗时:"+(System.currentTimeMillis()-beginTime));
	}
}

使用场景:http://www.jianshu.com/p/d3e5915a7ac5

本文转载自:http://xsh5324.iteye.com/blog/2058925?utm_source=tool.lu

chaun
粉丝 93
博文 269
码字总数 91059
作品 0
深圳
高级程序员
私信 提问

暂无文章

如何使用 rsync 备份 Linux 系统的一些介绍

备份一直是 Linux 世界的热门话题。回到 2017,David Both 为 Opensource.com 的读者在使用 rsync 备份 Linux 系统方面提了一些建议,在这年的更早时候,他发起了一项问卷调查询问大家,在 ...

xiangyunyan
32分钟前
0
0
二进制位操作

单片机,或者一些模块的设置操作,都是由一个字节数据来完成,每位各有定义。就需进行位操作来组合需要的数字结果。 以JavaScript为例,编写位操作。 我们期望得到这样一个二进制数:0101101...

format
45分钟前
3
0
聊聊中国的通信行业:从“七国八制”到“中华”脊梁

本期文章和大家一起来聊一聊我曾经从事过的通信行业吧。最近各方面信息的泛滥,包括和华为的同学聊天,自己确实也感慨颇多。想想我自己本科主修通信工程,研究生再修信息与通信工程,从本科开...

CodeSheep
今天
7
0
MDK:ARM M451M:exceed the range of code meory, continue to erase or not?

问题: 代码空间超限 几天前就遇到:exceed the range of code meory, continue to erase or not? 如下所示: 解决过程 开始以为中MDK软件的128KB限制,如是就不能生成HEX文件,应该链接时有提...

SamXIAO
今天
1
1
OSChina 周六乱弹 —— 因违反《中华人民共和国治安管理处罚法》第四十四条之规定

Osc乱弹歌单(2019)请戳(这里) 【今日歌曲】 @xiaoshiyue :#今日歌曲推荐# 惊艳分享谷微的单曲《安守本份》(@网易云音乐) 《安守本份》- 谷微 手机党少年们想听歌,请使劲儿戳(这里) ...

小小编辑
今天
653
14

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部