更多案例
使用原生 API 创建一个简单的生产者和消费者
//DEMO中使用的 消息全假定是一条交易
public class TradeTransaction {
private String id;//交易ID
private double 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();//终止线程
}
}
使用 WorkerPool 辅助创建消费者
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();
}
}
建立一个消费者的“四边形模式”，用 Disruptor 来完成整个构建工作
从中图可以看出需求是这样子的：生产者生产数据经过 C1,C2 处理完成后再到 C3，请忽略左下角的猫。假设如下场景：
- 交易网关收到交易(P1)把交易数据发到 RingBuffer 中，
- 负责处理增值业务的消费者 C1 和负责数据存储的消费者 C2 负责处理交易
- 负责发送 JMS 消息的消费者 C3 在 C1 和 C2 处理完成后再进行处理。
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));
}
}
多生产者、消费者
我们甚至可以在一个更复杂的六边形模式中构建一个并行消费者链：
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();
}
}
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);
}
}
}
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;
}
}
public class Main {
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 < 100; i++) {
final Producer p = new Producer(ringBuffer);
new Thread(new Runnable() {
@Override
public void run() {
try {
latch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
for(int j = 0; j < 100; j ++){
p.onData(UUID.randomUUID().toString());
}
}
}).start();
}
Thread.sleep(2000);
System.out.println("---------------开始生产-----------------");
latch.countDown();
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) {}
}
}
核心代码如下：
dw.consumeWith(handler1a, handler2a);
dw.after(handler1a).consumeWith(handler1b);
dw.after(handler2a).consumeWith(handler2b);
dw.after(handler1b, handler2b).consumeWith(handler3);
ProducerBarrier producerBarrier = dw.createProducerBarrier();