前往小程序,Get更优阅读体验!
立即前往
首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >Disruptor详解

Disruptor详解

作者头像
JavaEdge
发布2021-02-22 16:41:32
1.5K0
发布2021-02-22 16:41:32
举报
文章被收录于专栏:JavaEdge

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类:数据封装类

代码语言:javascript
复制
public class LongEvent {

	private Long value;

	public Long getValue() {
		return value;
	}

	public void setValue(Long value) {
		this.value = value;
	}
	
}

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

代码语言:javascript
复制
public class LongEventFactory implements EventFactory<LongEvent>{
	@Override
	public LongEvent newInstance() {
		return new LongEvent();
	}
}

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

代码语言:javascript
复制
/**
 * 消费者,事件监听
 * @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());
	}

}

数据生产类:

代码语言:javascript
复制
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);
		}
	}
}

测试类:

代码语言:javascript
复制
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消费类:

代码语言:javascript
复制
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数据封装类:

代码语言:javascript
复制
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测试类:

代码语言:javascript
复制
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测试类:

代码语言:javascript
复制
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:

代码语言:javascript
复制
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

代码语言:javascript
复制
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

代码语言:javascript
复制
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

代码语言:javascript
复制
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

代码语言:javascript
复制
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(生产者)

代码语言:javascript
复制
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:

代码语言:javascript
复制
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订单类:

代码语言:javascript
复制
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生产者:

代码语言:javascript
复制
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消费者:

代码语言:javascript
复制
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测试类:

代码语言:javascript
复制
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
本文参与 腾讯云自媒体同步曝光计划,分享自作者个人站点/博客。
原始发表:2020/10/11 ,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 作者个人站点/博客 前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体同步曝光计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
目录
  • Disruptor术语说明
  • 并行计算 - 多边形高端操作
    • 菱形操作
      • 六边形操作
      相关产品与服务
      GPU 云服务器
      GPU 云服务器(Cloud GPU Service,GPU)是提供 GPU 算力的弹性计算服务,具有超强的并行计算能力,作为 IaaS 层的尖兵利器,服务于生成式AI,自动驾驶,深度学习训练、科学计算、图形图像处理、视频编解码等场景。腾讯云随时提供触手可得的算力,有效缓解您的计算压力,提升业务效率与竞争力。
      领券
      问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档