Java异步批处理教程

书接上回 大数据量、高并发业务怎么优化?(一) 文章中介绍了异步批处理的三种方式,本文继续深入针对前两种进行讲解,并给出代码示例:

image.png

一 普通版本,采用阻塞队列 ArrayBlockingQueue

使用普通方式能够直接基于JDK中现成的并发包 ArrayBlockingQueue 提供的 offer(E e, long timeout, TimeUnit unit)(添加元素到队列尾部,如果队列已满则等待参数指定时间后返回false)方法 和 poll(long timeout, TimeUnit unit)(从队列头部获取元素,如果队列为空则等待参数指定时间后返回null)方法,来达到异步批处理效果

生产者代码:由于采用内存队列,最好在创建 ArrayBlockingQueue 时指定队列大小,防止队列无界,导致内存溢出

  • /**
  • * 生产者
  • */
  • @Component
  • @Slf4j
  • public class MonitorQueue {
  • private BlockingQueue<List<NodeCollectDTO>> queue = new ArrayBlockingQueue<>(10000000);
  • public void put(List<NodeCollectDTO> list) {
  • try {
  • queue.put(list);
  • } catch (InterruptedException e) {
  • log.error(String.format("队列put异常:%s", e.getMessage()), e);
  • }
  • }
  • public void offer(List<NodeCollectDTO> list, long timeout, TimeUnit unit) throws InterruptedException {
  • queue.offer(list, timeout, unit);
  • }
  • public List<NodeCollectDTO> poll(long timeout, TimeUnit unit) throws InterruptedException {
  • return queue.poll(timeout, unit);
  • }
  • }

消费者代码:在创建生产者时开启一个子线程在死循环中一直读取队列元素,直到队列元素超过我们的 maxNum 时,将临时列表元素插入数据库中

  • /**
  • * 消费者
  • */
  • @Slf4j
  • @Component
  • public class MonitorConsumer implements Runnable {
  • @Autowired
  • private MonitorQueue queue;
  • @Autowired
  • private MonitorService monitorService;
  • @PostConstruct
  • public void init() {
  • new Thread(this, "monitor-collect").start();
  • }
  • // 临时列表大小限制
  • private int maxNum = 2000;
  • @SuppressWarnings("InfiniteLoopStatement")
  • @Override
  • public void run() {
  • while (true) {
  • handler();
  • }
  • }
  • private void handler() {
  • try {
  • List<NodeCollectDTO> temp = new ArrayList<>(maxNum);
  • while (temp.size() <= maxNum) {
  • List<NodeCollectDTO> list = queue.poll(20, TimeUnit.SECONDS);
  • if (CollectionUtil.isNotEmpty(list)) {
  • temp.addAll(list);
  • } else {
  • break;
  • }
  • }
  • if (CollectionUtil.isEmpty(temp)) {
  • return;
  • }
  • int i = monitorService.batchSave(temp);
  • log.debug("----------------------------batchSave num:{}, collect.size:{}", i, collect.size());
  • } catch (Exception e) {
  • log.error(String.format("消费者异常: %s", e.getMessage()), e);
  • }
  • }
  • }
展开

可以看到采用该种方式实现的异步批量入库代码比较简单,便于理解,在性能上,基本都能够满足日常普通业务存在的批量入库场景

二 进阶版,采用 Disruptor 队列,本文基于 Disruptor 最新4.0版本

先给出 Disruptor 官网简介

Disruptor 是一个提供并发环形缓冲区数据结构的库。它旨在在异步事件处理架构中提供低延迟、高吞吐量的工作队列。
为了理解 Disruptor 的好处,我们可以将它与一些很好理解且目的非常相似的东西进行比较。在 Disruptor 的情况下,这将是 Java 的 BlockingQueue。与队列一样,Disruptor 的目的是在同一进程内的线程之间移动数据(例如消息或事件)。然而,Disruptor 提供的一些关键特性使其有别于队列。他们是:

向消费者多播事件,带有消费者依赖图。

为事件预分配内存。

可选无锁

Disruptor 给我们在项目中实现异步批处理提供了另一种方式,一种无锁、延迟更低、吞吐量更高、提供消费者多播等等的内存队列

下面介绍如何使用

2.1 依赖安装

  • <dependency>
  • <groupId>com.lmax</groupId>
  • <artifactId>disruptor</artifactId>
  • <version>4.0.0.RC1</version>
  • </dependency>

2.2 Disruptor 使用代码如下:

  • public class LongEvent{
  • private long value;
  • public void set(long value){
  • this.value = value;
  • }
  • @Override
  • public String toString(){
  • return "LongEvent{" + "value=" + value + '}';
  • }
  • }
  • @Slf4j
  • public class LongEventMain {
  • public static void handleEvent(LongEvent event, long sequence, boolean endOfBatch) {
  • log.info("event: " + event + ", sequence:" + sequence + ", endOfBatch:" + endOfBatch);
  • }
  • public static void translate(LongEvent event, long sequence, ByteBuffer buffer) {
  • event.set(buffer.getLong(0));
  • }
  • public static void main(String[] args) throws Exception {
  • int bufferSize = 128;
  • // 1. 创建Disruptor对象
  • Disruptor<LongEvent> disruptor =
  • new Disruptor<>(LongEvent::new, bufferSize, DaemonThreadFactory.INSTANCE, ProducerType.SINGLE, new BlockingWaitStrategy());
  • // 2. 添加事件处理类(消费者)
  • disruptor.handleEventsWith(LongEventMain::handleEvent);
  • // 3. 开启事件处理线程
  • disruptor.start();
  • // 4. 获取ringBuffer
  • RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();
  • ByteBuffer bb = ByteBuffer.allocate(8);
  • for (long l = 0; true; l++) {
  • bb.putLong(0, l);
  • // 5. 发布事件(生产者)
  • ringBuffer.publishEvent(LongEventMain::translate, bb);
  • Thread.sleep(1);
  • }
  • }
  • }
展开

2.3 上面代码完成了一个事件发布后,事件处理类就能够收到对应事件信息的功能,但是我们想要的是能在消费者线程中批量处理生产者数据的逻辑,还得再修改一下事件处理类代码,如下:

  • @Slf4j
  • public class LongEventBatch implements EventHandler<LongEvent> {
  • private static final int MAX_BATCH_SIZE = 20;
  • private final List<LongEvent> batch = new ArrayList<>();
  • public LongEventBatch() {
  • // 虚拟机关闭处理
  • Runtime.getRuntime().addShutdownHook(new Thread(() -> {
  • log.info("------------------ShutdownHook-DataEventHandler,上报tempList");
  • if (batch.size() > 0) {
  • // 批量入库伪代码
  • int i = xxxService.batchSave(temp);
  • }
  • }));
  • }
  • @Override
  • public void onEvent(final LongEvent event, final long sequence, final boolean endOfBatch) {
  • log.info("event: " + event + ", sequence:" + sequence + ", endOfBatch:" + endOfBatch);
  • batch.add(event);
  • if (batch.size() >= MAX_BATCH_SIZE) {
  • processBatch(batch);
  • }
  • }
  • private void processBatch(final List<LongEvent> batch) {
  • // 批量入库伪代码
  • int i = xxxService.batchSave(temp);
  • // 记得清空batch列表
  • batch.clear();
  • }
  • }

由此,我们就实现了基于 Disruptor 的异步批处理逻辑,该方式会比普通版本性能高出一个数量级,大家在工作中可以尝试使用一番

最后

附博主 github 地址 https://github.com/wayn111

本站文章资源均来源自网络,除非特别声明,否则均不代表站方观点,并仅供查阅,不作为任何参考依据!
如有侵权请及时跟我们联系,本站将及时删除!
如遇版权问题,请查看 本站版权声明
THE END
分享
二维码
海报
<<上一篇
下一篇>>