并发框架disruptor（高性能内存Queue）

public long tryNext(int n) throws InsufficientCapacityException
 {
     if (n  0");
     }
  
     long current;
     long next;
  
     do
     {
         current = cursor.get();
         next = current + n;
  
         if (!hasAvailableCapacity(gatingSequences, n, current))
         {
             throw InsufficientCapacityException.INSTANCE;
         }
     }
     while (!cursor.compareAndSet(current, next));
  
     return next;
 }

package com.vrv.linkdood_disruptor;
  
 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 java.util.concurrent.TimeoutException;
  
 import com.lmax.disruptor.BatchEventProcessor;
 import com.lmax.disruptor.EventFactory;
 import com.lmax.disruptor.RingBuffer;
 import com.lmax.disruptor.SequenceBarrier;
 import com.lmax.disruptor.YieldingWaitStrategy;
  
 /**
  *      说    明:开始
  * 
  * @author 作者名：冯龙淼
  *            E-mail：fenglongmiao@vrvmail.com.cn
  * 
  * @version 版   本  号：1.0.

  *          创建时间：2018年1月22日 下午12:12:31
  */
 public class DemoMain {  
     
     public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {  
         
         final int BUFFER_SIZE=1024;  
         final int THREAD_NUMBERS=4;  
         
         /* 
          * createSingleProducer创建一个单生产者的RingBuffer， 
          * 第一个参数叫EventFactory，从名字上理解就是“事件工厂”，其实它的职责就是产生数据填充RingBuffer的区块。 
          * 第二个参数是RingBuffer的大小，它必须是2的指数倍 目的是为了将求模运算转为&运算提高效率 
          * 第三个参数是RingBuffer的生产都在没有可用区块的时候(可能是消费者（或者说是事件处理器） 太慢了)的等待策略 
          */  
         final RingBuffer ringBuffer = RingBuffer.createSingleProducer(new EventFactory() {  
             @Override  
             public TradeTransaction newInstance() {  
                 return new TradeTransaction();  
             } // PhasedBackoffWaitStrategy YieldingWaitStrategy
         }, BUFFER_SIZE,new YieldingWaitStrategy());  
         //创建线程池  
         ExecutorService executors = Executors.newFixedThreadPool(THREAD_NUMBERS);  
         //创建SequenceBarrier  
         SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();  
           
         //创建消息处理器  
         BatchEventProcessor transProcessor = new BatchEventProcessor(  
                 ringBuffer, sequenceBarrier, new TradeTransactionInDBHandler());  
           
         //这一部的目的是让RingBuffer根据消费者的状态    如果只有一个消费者的情况可以省略  
         ringBuffer.addGatingSequences(transProcessor.getSequence());  
           
         //把消息处理器提交到线程池  
         executors.submit(transProcessor);  
         //如果存大多个消费者 那重复执行上面3行代码 把TradeTransactionInDBHandler换成其它消费者类  
           
         Future> future=executors.submit(new Callable() {  
             @Override  
             public Void call() throws Exception {  
                 long seq = 0;  
                 for(int i=0;i说    明:数据实体类
  * 
  * @author 作者名：冯龙淼
  *            E-mail：fenglongmiao@vrvmail.com.cn
  * 
  * @version 版   本  号：1.0.

  *          创建时间：2018年1月22日 下午12:10:58
  */
 //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;  
    }
     @Override
     public String toString() {
         return "TradeTransaction [id=" + id + ", price=" + price + "]";
     }  
  
 }  package com.vrv.linkdood_disruptor;
  
 import java.util.UUID;
 import java.util.concurrent.atomic.AtomicLong;
  
 import com.lmax.disruptor.EventHandler;
 import com.lmax.disruptor.WorkHandler;
  
 /**
  *      说    明:消费实现类
  * 
  * @author 作者名：冯龙淼
  *            E-mail：fenglongmiao@vrvmail.com.cn
  * 
  * @version 版   本  号：1.0.

  *          创建时间：2018年1月22日 下午12:11:33
  */
 public class TradeTransactionInDBHandler implements EventHandler,WorkHandler {  
     
     /** 交易条数 */
     private AtomicLong count = new AtomicLong(0);
     
     @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()+"&& price:"+event.getPrice()+"count: "+count.getAndIncrement()); 
     }  
     
 }

try{
  }finally{
  },

public class LongEvent
 {
     private long value;
  
     public void set(long value)
     {
         this.value = value;
     }
 }
 static class Translator implements EventTranslatorOneArg{
     @Override
     publicvoid translateTo(LongEvent event, long sequence, Long data) {
         event.set(data);
     }    
 }
     
 public static Translator TRANSLATOR = new Translator();
     
 public staticvoid publishEvent2(Disruptor disruptor) {
     // 发布事件；
     RingBuffer ringBuffer = disruptor.getRingBuffer();
     long data = getEventDataxxxx();//获取要通过事件传递的业务数据；    
     ringBuffer.publishEvent(TRANSLATOR,data);
 }

class ComsumerHandler implements EventHandler{
     
     private int no;
  
     public  ComsumerHandler(int no) {
         this.no=no;
     }
  
     @Override
     public void onEvent(ResultTxt resultTxt, long sequence, boolean endOfBatch)
         throws Exception {
         System.out.println(no+" data commming......"+resultTxt.getBarCode());
     }
 }
  
 //设置多个消费者
 disruptor.handleEventsWith(new ComsumerHandler(1),new ComsumerHandler(2));
      （类型2）  多个消费者，每个消费者消费不同数据。也就是说每个消费者竞争数据，竞争到消费，其他消费者没有机会。使用handleEventsWithWorkerPool方法class ComsumerHandler implements WorkHandler{
  
     private int no;
  
     public  ComsumerHandler(int no) {
         this.no=no;
     }
  
     @Override
     public void onEvent(ResultTxt event)
         throws Exception {
         System.out.println(no+"  data commming......"+event.getBarCode());
     }
 }
  
 //多个消费者，每个消费者竞争消费不同数据
 disruptor.handleEventsWithWorkerPool(new ComsumerHandler(1),new ComsumerHandler(2));