Disrunptor多生产者多消费者模型讲解

package com.ljq.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;
	}

}

package com.ljq.multi;

import com.lmax.disruptor.RingBuffer;

/**
 * 生产者
 * 
 * @author Administrator
 *
 */
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);
		}
	}

}

package com.ljq.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 {
		count.incrementAndGet();
		System.out.println("当前消费者: " + this.consumerId + "，消费信息：" + order.getId() + "，count：" + getCount());
	}
	
	public int getCount(){
		return count.get();
	}

}

package com.ljq.multi;

import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

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.ProducerType;

/**
 * 	disrunptor 多生产者多消费者
 *  
 * @author Administrator
 *
 */
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();
		
		//创建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<Order>(ringBuffer, 
						barriers, 
						new IntEventExceptionHandler(),
						consumers);
		
        ringBuffer.addGatingSequences(workerPool.getWorkerSequences());  
        workerPool.start(Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors()));  
        
      
        final AtomicInteger count = new AtomicInteger(0);
        final CountDownLatch latch = new CountDownLatch(1);
        //100个生产者
        for (int i = 0; i < 100; i++) {  
        	final Producer p = new Producer(ringBuffer); 
        	new Thread(new Runnable() {
				@Override
				public void run() {
					try {
						System.out.println(count.incrementAndGet());
						latch.await(); //等待，直到100个生产者创建完成，再一起生产消费订单数据
						
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
					
					//每个生产者生产100条订单
					for(int j = 0; j < 100; j ++){
						p.onData(UUID.randomUUID().toString());
					}
				}
			}).start();
        } 
        System.out.println("---------------开始生产-----------------");
  
        latch.countDown(); //生产者全部创建完毕，开始生产订单
        
	}
	
	static class IntEventExceptionHandler implements ExceptionHandler {  
	    public void handleEventException(Throwable ex, long sequence, Object event) {}  
	    public void handleOnStartException(Throwable ex) {}  
	    public void handleOnShutdownException(Throwable ex) {}  
	} 
}