并发编程之Disruptor框架介绍和高阶运用（二）

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.YieldingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.EventHandlerGroup;
import com.lmax.disruptor.dsl.ProducerType;
/**
 * 测试 P1生产消息，C1，C2消费消息，C1和C2会共享所有的event元素! C3依赖C1，C2处理结果
 * @author lzhcode
 *
 */
public class Main {
  public static void main(String[] args) throws InterruptedException {
    long beginTime = System.currentTimeMillis();
    //最好是2的n次方
    int bufferSize = 1024;
    // Disruptor交给线程池来处理，共计 p1,c1,c2,c3四个线程
    ExecutorService executor = Executors.newFixedThreadPool(4);
    // 构造缓冲区与事件生成
    Disruptor<InParkingDataEvent> disruptor = new Disruptor<InParkingDataEvent>(
        new EventFactory<InParkingDataEvent>() {
          @Override
          public InParkingDataEvent newInstance() {
            return new InParkingDataEvent();
          }
        }, bufferSize, executor, ProducerType.SINGLE, new YieldingWaitStrategy());
    // 使用disruptor创建消费者组C1,C2
    EventHandlerGroup<InParkingDataEvent> handlerGroup = disruptor.handleEventsWith(new ParkingDataToKafkaHandler(),
        new ParkingDataInDbHandler());
    ParkingDataSmsHandler smsHandler = new ParkingDataSmsHandler();
    // 声明在C1,C2完事之后执行JMS消息发送操作 也就是流程走到C3
    handlerGroup.then(smsHandler);
    disruptor.start();// 启动
    CountDownLatch latch = new CountDownLatch(1);
    // 生产者准备
    executor.submit(new InParkingDataEventPublisher(latch, disruptor));
    latch.await();// 等待生产者结束
    disruptor.shutdown();
    executor.shutdown();
    System.out.println("总耗时:" + (System.currentTimeMillis() - beginTime));
  }
}

import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
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;
/**
 * 多生产者多消费者模型
 * @author lzhcode
 *
 */
public class Main {
  public static void main(String[] args) throws InterruptedException {
    //1 创建RingBuffer
    RingBuffer<Order> ringBuffer =
        RingBuffer.create(ProducerType.MULTI,
            new EventFactory<Order>() {
              public Order newInstance() {
                return new Order();
              }
            },
            1024*1024,
            new YieldingWaitStrategy());
    //2 通过ringBuffer 创建一个屏障
    SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();
    //3 创建一个消费者:
    Consumer consumer = new Consumer;
    //4 构建多消费者工作池
    WorkerPool<Order> workerPool = new WorkerPool<Order>(
        ringBuffer,
        sequenceBarrier,
        new EventExceptionHandler(),
        consumer );
    //5 设置消费者的sequence序号 用于单独统计消费进度, 并且设置到ringbuffer中
    ringBuffer.addGatingSequences(workerPool.getWorkerSequences());
    //6 启动workerPool
    workerPool
    .start(Executors.newFixedThreadPool(5));
    final CountDownLatch latch = new CountDownLatch(1);
    for(int i = 0; i < 100; i++) {
      final Producer producer = new Producer(ringBuffer);
      new Thread(new Runnable() {
        public void run() {
          try {
            latch.await();
          } catch (Exception e) {
            e.printStackTrace();
          }
          for(int j = 0; j<100; j++) {
            producer.sendData(UUID.randomUUID().toString());
          }
        }
      }).start();
    }
    Thread.sleep(2000);
    System.err.println("----------线程创建完毕，开始生产数据----------");
    latch.countDown();
    Thread.sleep(10000);
    System.err.println("任务总数:" + consumers[2].getCount());
  }
  static class EventExceptionHandler implements ExceptionHandler<Order> {
    public void handleEventException(Throwable ex, long sequence, Order event) {
    }
    public void handleOnStartException(Throwable ex) {
    }
    public void handleOnShutdownException(Throwable ex) {
    }
  }
}