Disruptor3.2实现一个生产者与两个消费者

项目的目录结构如下：

 

1.定义事件

package org.fenxisoft.disruptor;

//定义事件
public class LongEvent {
	
	private long value;
	
	public void set(long value){
		this.value = value;
	}
    
	@Override
	public String toString() {
		return String.valueOf(value);
	}
}

 

 

2.定义事件工厂

package org.fenxisoft.disruptor;

import com.lmax.disruptor.EventFactory;

public class LongEventFactory implements EventFactory<LongEvent>{

	@Override
	public LongEvent newInstance() {
		return new LongEvent();
	}

}

 

3.定义事件处理者

package org.fenxisoft.disruptor;

import com.lmax.disruptor.EventHandler;

public class LongEventHandler implements EventHandler<LongEvent>{

	@Override
	public void onEvent(LongEvent event, long sequence, boolean endOfBatch)
			throws Exception {
		System.out.println(Thread.currentThread().getName()+" : "+event);
		
	}

}

4.编写测试类

package org.fenxisoft.disruptor;

import java.util.concurrent.Executors;

import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.IgnoreExceptionHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.Sequence;
import com.lmax.disruptor.WaitStrategy;
import com.lmax.disruptor.WorkHandler;
import com.lmax.disruptor.WorkerPool;
import com.lmax.disruptor.dsl.ProducerType;

public class DisruptorDemo {
	
	public static void main(String[] args) {
		
		LongEventFactory eventFactory = new LongEventFactory();		
		int bufferSize = 4;//1024 * 1024;//ring Buffer Size
		WaitStrategy waitStrategy = new BlockingWaitStrategy();
		
		RingBuffer<LongEvent> ringBuffer = RingBuffer.create(ProducerType.MULTI, eventFactory,
				bufferSize, waitStrategy);
		
		WorkHandler<LongEvent> workHandler1 = new WorkHandler<LongEvent>() {

			@Override
			
			public void onEvent(LongEvent event) throws Exception {
				
				System.out.println(Thread.currentThread().getName()+"消费数据： "+event);
			}
		};
		
		
		WorkerPool<LongEvent> workerPool = new WorkerPool<LongEvent>(ringBuffer, 
				ringBuffer.newBarrier(), new IgnoreExceptionHandler(), workHandler1,workHandler1);
		
		//每个消费者，也就是workProcessor都有一个sequence
		Sequence[] sequences = workerPool.getWorkerSequences();
		ringBuffer.addGatingSequences(sequences);
		workerPool.start(Executors.newFixedThreadPool(10));
		
		System.out.println("运行的线程数量："+Thread.activeCount());//由此可以说明生产者就是主线程，消费者是两个子线程
		//生产者生产数据
		System.out.println("开始生产");
		for(int i = 0; i < 100; i++){
			long sequence = ringBuffer.next();
			
			try {
				LongEvent event = ringBuffer.get(sequence);
				event.set(i);
			} finally{
				ringBuffer.publish(sequence);
			}
			System.out.println("-----------------");
			
		}
		
	}
	
	

}

 

核心方法分析：

1.next()方法分析，这个方法是单线程调用

  public long next(int n)
    {
        if (n < 1)
        {
            throw new IllegalArgumentException("n must be > 0");
        }

        long current;//定义当前坐标
        long next;//定义下次要访问的数组坐标

        do
        {
        	current = cursor.get();//获取当前坐标
            next = current + n;//当前坐标加1就是下次要访问的数组坐标
            
            long wrapPoint = next - bufferSize;
            long cachedGatingSequence = gatingSequenceCache.get();
            
            //如果成立，说明生产过快
            if (wrapPoint > cachedGatingSequence || cachedGatingSequence > current)
            {
            	//获取已经消费【或者准备消费的】的最小序列
                long gatingSequence = Util.getMinimumSequence(gatingSequences, current);
                //如果成立，说明需要等待消费者消费事件
                if (wrapPoint > gatingSequence)
                {
                    waitStrategy.signalAllWhenBlocking();//尝试唤醒消费者WorkProcessor来消费事件Event
                    LockSupport.parkNanos(1); // TODO, should we spin based on the wait strategy?
                    continue;
                }

                gatingSequenceCache.set(gatingSequence);
            }
            else if (cursor.compareAndSet(current, next))
            {
                break;
            }
        }
        while (true);

        return next;
    }

 

2.run()方法分析，这个方法是多线程调用

 public void run()
    {
        if (!running.compareAndSet(false, true))
        {
            throw new IllegalStateException("Thread is already running");
        }
        sequenceBarrier.clearAlert();

        notifyStart();

        boolean processedSequence = true;
        long cachedAvailableSequence = Long.MIN_VALUE;
        long nextSequence = sequence.get();
        T event = null;
        while (true)
        {
            try
            {
                // if previous sequence was processed - fetch the next sequence and set
                // that we have successfully processed the previous sequence
                // typically, this will be true
                // this prevents the sequence getting too far forward if an exception
                // is thrown from the WorkHandler
                if (processedSequence)
                {
                    processedSequence = false;
                    do
                    {
                        nextSequence = workSequence.get() + 1L;
                        sequence.set(nextSequence - 1L);
                    }
                    //两个WorkProcessor共用同一个workSequence,所以不会出现同一个事件Event被2个消费者消费
                    while (!workSequence.compareAndSet(nextSequence - 1L, nextSequence));
                }
                if (cachedAvailableSequence >= nextSequence)
                {
                    event = ringBuffer.get(nextSequence);
                    workHandler.onEvent(event);
                    processedSequence = true;
                }
                else
                {    
                	//获取可以消费的最大序列
                    cachedAvailableSequence = sequenceBarrier.waitFor(nextSequence);
                }
            }
            catch (final TimeoutException e)
            {
                notifyTimeout(sequence.get());
            }
            catch (final AlertException ex)
            {
                if (!running.get())
                {
                    break;
                }
            }
            catch (final Throwable ex)
            {
                // handle, mark as processed, unless the exception handler threw an exception
                exceptionHandler.handleEventException(ex, nextSequence, event);
                processedSequence = true;
            }
        }

        notifyShutdown();

        running.set(false);
    }

 

 

如果没明白，看下图：

 

 

 参考：http://www.cnblogs.com/jony-zhang/p/3817208.html

           http://maoyidao.iteye.com/blog/1663193