并发框架disruptor_ringbuffer的常规用法

import java.nio.ByteBuffer;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import org.apache.log4j.LogManager;
import org.apache.logging.log4j.core.Logger;
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();
//创建工厂
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
Disruptor<LongEvent> disruptor = 
new Disruptor<LongEvent>(factory, ringBufferSize, executor, ProducerType.SINGLE, new YieldingWaitStrategy());
// 连接消费事件方法
disruptor.handleEventsWith(new LongEventHandler());
// 启动
disruptor.start();
//Disruptor 的事件发布过程是一个两阶段提交的过程：
//发布事件
RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();
org.apache.log4j.Logger logger1 = LogManager.getLogger(LongEventMain.class);
LongEventProducer producer = new LongEventProducer(ringBuffer); 
//LongEventProducerWithTranslator producer = new LongEventProducerWithTranslator(ringBuffer);
ByteBuffer byteBuffer = ByteBuffer.allocate(8);
for(long l = 0; l<1000; l++){
byteBuffer.putLong(0, l);
producer.onData(byteBuffer);
//Thread.sleep(1000);
}
disruptor.shutdown();//关闭 disruptor，方法会堵塞，直至所有的事件都得到处理；
executor.shutdown();//关闭 disruptor 使用的线程池；如果需要的话，必须手动关闭， disruptor 在 shutdown 时不会自动关闭；        
}
}

import com.lmax.disruptor.EventFactory;
//需要让disruptor为我们创建事件，我们同时还声明了一个EventFactory来实例化Event对象。
public class LongEventFactory implements EventFactory { 
public Object newInstance() { 
return new LongEvent(); 
} 
} 

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

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

//我们还需要一个事件消费者，也就是一个事件处理器。这个事件处理器简单地把事件中存储的数据打印到终端：
public class LongEventHandler implements EventHandler<LongEvent>  {
public void onEvent(LongEvent longEvent, long l, boolean b) throws Exception {
System.out.println(longEvent.getValue()+" haha");         
}
}