架构师入门笔记八 并发框架Disruptor场景应用

架构师入门笔记八 并发框架Disruptor场景应用 今天用一个停车场问题来加深对Disruptor的理解。一个有关汽车进入停车场的问题。 当汽车进入停车场时,系统首先会记录汽车信息。同时也会发送消息到其他系统处理相关业务,最后发送短信通知车主收费开始。看了很多文章,里面的代码都是大同小异的,可能代码真的是很经典。以下代码也是来源网络,只是自己手动敲的,加了一些注释。
代码包含以下内容: 1) 事件对象Event 2)三个消费者Handler 3)一个生产者Processer 4)执行Main方法 Event类:汽车信息
public class MyInParkingDataEvent {

	private String carLicense; // 车牌号

	public String getCarLicense() {
		return carLicense;
	}

	public void setCarLicense(String carLicense) {
		this.carLicense = carLicense;
	}

}
Handler类:一个负责存储汽车数据,一个负责发送kafka信息到其他系统中,最后一个负责给车主发短信通知
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.WorkHandler;

/**
 * Handler 第一个消费者,负责保存进场汽车的信息
 *
 */
public class MyParkingDataInDbHandler implements EventHandlerMyInParkingDataEvent , WorkHandlerMyInParkingDataEvent{

	@Override
	public void onEvent(MyInParkingDataEvent myInParkingDataEvent) throws Exception {
		long threadId = Thread.currentThread().getId(); // 获取当前线程id
		String carLicense = myInParkingDataEvent.getCarLicense(); // 获取车牌号
		System.out.println(String.format("Thread Id %s 保存 %s 到数据库中 ....", threadId, carLicense));
	}

	@Override
	public void onEvent(MyInParkingDataEvent myInParkingDataEvent, long sequence, boolean endOfBatch)
			throws Exception {
		this.onEvent(myInParkingDataEvent);
	}

}
import com.lmax.disruptor.EventHandler;

/**
 * 第二个消费者,负责发送通知告知工作人员(Kafka是一种高吞吐量的分布式发布订阅消息系统)
 */
public class MyParkingDataToKafkaHandler implements EventHandlerMyInParkingDataEvent{

	@Override
	public void onEvent(MyInParkingDataEvent myInParkingDataEvent, long sequence, boolean endOfBatch)
			throws Exception {
		long threadId = Thread.currentThread().getId(); // 获取当前线程id
		String carLicense = myInParkingDataEvent.getCarLicense(); // 获取车牌号
		System.out.println(String.format("Thread Id %s 发送 %s 进入停车场信息给 kafka系统...", threadId, carLicense));
	}

}
import com.lmax.disruptor.EventHandler;

/**
 * 第三个消费者,sms短信服务,告知司机你已经进入停车场,计费开始。
 */
public class MyParkingDataSmsHandler implements EventHandlerMyInParkingDataEvent{

	@Override
	public void onEvent(MyInParkingDataEvent myInParkingDataEvent, long sequence, boolean endOfBatch)
			throws Exception {
		long threadId = Thread.currentThread().getId(); // 获取当前线程id
		String carLicense = myInParkingDataEvent.getCarLicense(); // 获取车牌号
		System.out.println(String.format("Thread Id %s 给  %s 的车主发送一条短信,并告知他计费开始了 ....", threadId, carLicense));
	}

}
Producer类:负责上报停车数据
import java.util.concurrent.CountDownLatch;
import com.lmax.disruptor.EventTranslator;
import com.lmax.disruptor.dsl.Disruptor;

/**
 * 生产者,进入停车场的车辆
 */
public class MyInParkingDataEventPublisher implements Runnable{
	
	private CountDownLatch countDownLatch; // 用于监听初始化操作,等初始化执行完毕后,通知主线程继续工作
	private DisruptorMyInParkingDataEvent disruptor;
	private static final Integer NUM = 1; // 1,10,100,1000
	
	public MyInParkingDataEventPublisher(CountDownLatch countDownLatch,
			DisruptorMyInParkingDataEvent disruptor) {
		this.countDownLatch = countDownLatch;
		this.disruptor = disruptor;
	}
	
	@Override
	public void run() {
		MyInParkingDataEventTranslator eventTranslator = new MyInParkingDataEventTranslator();
		try {
			for(int i = 0; i  NUM; i ++) {
				disruptor.publishEvent(eventTranslator);
				Thread.sleep(1000); // 假设一秒钟进一辆车
			}
		} catch (InterruptedException e) {
			e.printStackTrace();
		} finally {
			countDownLatch.countDown(); // 执行完毕后通知 await()方法
			System.out.println(NUM + "辆车已经全部进入进入停车场!");
		}
	}
	
}

class MyInParkingDataEventTranslator implements EventTranslatorMyInParkingDataEvent {

	@Override
	public void translateTo(MyInParkingDataEvent myInParkingDataEvent, long sequence) {
		this.generateData(myInParkingDataEvent);
	}
	
	private MyInParkingDataEvent generateData(MyInParkingDataEvent myInParkingDataEvent) {
		myInParkingDataEvent.setCarLicense("车牌号: 鄂A-" + (int)(Math.random() * 100000)); // 随机生成一个车牌号
		System.out.println("Thread Id " + Thread.currentThread().getId() + " 写完一个event");
		return myInParkingDataEvent;
	}
	
}
执行的Main方法:
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;

/**
 * 执行的Main方法 ,
 * 一个生产者(汽车进入停车场);
 * 三个消费者(一个记录汽车信息,一个发送消息给系统,一个发送消息告知司机)
 * 前两个消费者同步执行,都有结果了再执行第三个消费者
 */
public class MyInParkingDataEventMain {
	
	public static void main(String[] args) {
		long beginTime=System.currentTimeMillis();
		int bufferSize = 2048; // 2的N次方
		try {
			// 创建线程池,负责处理Disruptor的四个消费者
			ExecutorService executor = Executors.newFixedThreadPool(4);
			
			// 初始化一个 Disruptor
			DisruptorMyInParkingDataEvent disruptor = new DisruptorMyInParkingDataEvent(new EventFactoryMyInParkingDataEvent() {
				@Override
				public MyInParkingDataEvent newInstance() {
					return new MyInParkingDataEvent(); // Event 初始化工厂
				}
			}, bufferSize, executor, ProducerType.SINGLE, new YieldingWaitStrategy());
			
			// 使用disruptor创建消费者组 MyParkingDataInDbHandler 和 MyParkingDataToKafkaHandler
			EventHandlerGroupMyInParkingDataEvent handlerGroup = disruptor.handleEventsWith(
					new MyParkingDataInDbHandler(), new MyParkingDataToKafkaHandler());
			
			// 当上面两个消费者处理结束后在消耗 smsHandler
			MyParkingDataSmsHandler myParkingDataSmsHandler = new MyParkingDataSmsHandler();
			handlerGroup.then(myParkingDataSmsHandler);
			
			// 启动Disruptor
			disruptor.start();
			
			CountDownLatch countDownLatch = new CountDownLatch(1); // 一个生产者线程准备好了就可以通知主线程继续工作了
			// 生产者生成数据
			executor.submit(new MyInParkingDataEventPublisher(countDownLatch, disruptor));
			countDownLatch.await(); // 等待生产者结束
			
			disruptor.shutdown();
			executor.shutdown();
		} catch (Exception e) {
			e.printStackTrace();
		}
		
		System.out.println("总耗时:"+(System.currentTimeMillis()-beginTime));
	}
	
}




用了两篇博客简单的介绍了Disruptor并发框架,如果想深入学习,可以到并发网里面找文章。下一章介绍BIO,NIO,AIO知识,为Netty5的入门打个基础。

学习博客:
LMAX Disruptor——一个高性能、低延迟且简单的框架
简单了解Disruptor


最新回复(0)
/jishuZKVGCJgmPgvXLMXYVUjvOroX9oSvt95NgxV4oKuLYUA_3D4795039
8 简首页