大家好，我是君哥。
使用消息队列时，为了提高生产和消费的性能，有时会开启批量处理。
在生产端，生产者发送的消息先发送到一个消息列表，积累到一定的消息量之后再批量发送给 Broker，如下图：
在消费端，消费者拉取消息后先不立即处理，而是把消息转存到一个内存队列或数据库，由业务线程去处理，如下图：
无论是生产者做批量发送，还是消费者做批量处理，都需要考虑使用批量消息的业务场景，避免踩坑。下面看一下批量操作可能会遇到哪些坑。
当生产者采用批量发送的方式来提高发送性能时，一定要考虑发送消息的批量大小。下面是 RocketMQ 批量发送的官方示例：
String topic = "BatchTest";List<Message> messages = new ArrayList<>();messages.add(new Message(topic, "TagA", "OrderID001", "Hello world 0".getBytes()));messages.add(new Message(topic, "TagA", "OrderID002", "Hello world 1".getBytes()));messages.add(new Message(topic, "TagA", "OrderID003", "Hello world 2".getBytes()));try { producer.send(messages);} catch (Exception e) { e.printStackTrace(); //handle the error}RocketMQ 默认消息大小是 4M，由 maxMessageSize 参数控制，如果批量消息大小超过 maxMessageSize，则会抛出异常。
如果遇到消息大小超过 maxMessageSize 的情况时，可以用下面方法进行处理：
RabbitMQ 相关的 API 则提供了更加灵活的批量控制，对消息数量和消息大小都做了控制，下面看一下源码：
消费端可以批量拉取消息进行消费，这样可以减少拉取消息时的 RPC 次数，提升消费性能。比如在 RocketMQ 中，可以通过 Consumer 中的 pullBatchSize 来设置一次拉取的消息数量，通过 consumeMessageBatchMaxSize 参数来设置一次消费的消息数量。
但需要注意的是，如果批量消息中一条消息消费失败了，这一批消息都需要进行重试，已经消费成功的消息会被重复消费，带来业务问题。
为了不对业务造成影响，必须考虑幂等。一个简单的方法是在消息中增加全局唯一 id 属性，对消息消费结果进行记录，消费成功后保存 id。这样在消费消息之前先查询是否存在消费成功的记录，如果存在则直接返回处理成功。
在使用消息队列进行批量操作时，必须要考虑到时延问题。比如我们设置一个批次 100 条消息，积累够 100 条消息后再发送，在消息量小的情况下，可能积累够 100 条消息会很长时间，导致消费端拉取到一条消息时延很大。
虽然消息队列的一个重要作用是削峰填谷，但在一些场景下，对消息的实时性也有要求。比如在车联网的充电场景，车联网平台需要实时感知充电桩的状态，如果充电桩积累够一批消息再上报平台，平台获取到的状态会不准确，如果心跳消息延时太久，平台会认为充电桩离线。
对于有时延要求又需要批量操作的场景，可以设置一个超时时间，超时后即使消息数量不够，也会发送出去。看下 RabbitMQ 的处理：
public synchronized void send(String exchange, String routingKey, Message message, CorrelationData correlationData) throws AmqpException { if (correlationData != null) { //... super.send(exchange, routingKey, message, correlationData); } else { if (this.scheduledTask != null) { this.scheduledTask.cancel(false); } MessageBatch batch = this.batchingStrategy.addToBatch(exchange, routingKey, message); if (batch != null) { super.send(batch.getExchange(), batch.getRoutingKey(), batch.getMessage(), null); } //这里获取到超时时间，到达超时时间后使用定时器将消息发送出去 Date next = this.batchingStrategy.nextRelease(); if (next != null) { this.scheduledTask = this.scheduler.schedule((Runnable) () -> releaseBatches(), next); } }}使用批处理一定要考虑可靠性的问题。
在消费端，消费者批量拉取一批消息后把消息暂存到一个内存临时队列，然后多线程去临时队列消费消息，如果服务宕机，临时队列中的消息会丢失。
为了避免宕机引发的损失，可以拉取一批消息后保存到数据库，然后给 Broker 返回 ACK，之后业务代码去数据库查询消息并消费，不过要考虑数据库大事务、锁竞争等问题。
当然，对于一些消息丢失不敏感的场景，比如日志收集之类的，可靠性这个指标是不用太关注的。
因为批量消息有一些复杂性，消息队列的部分特性不支持。
批量消息会增加消息重试的难度，所以对于事务消息，建议使用单条消息，一条消息对应一个事务。
顺序消息的实现思路一般是生产者将消息发送到同一个分区，消费者绑定这个分区并使用单线程消费这个分区的消息。如果对同一个 Topic 下的同一个分区来实现批量发送，难度会增大。所以建议顺序消息使用单条消息进行发送。
如果延时消息使用批量进行发送，这一批消息的延时时间必须相同，同时要考虑批量消息的超时时间，超时时间太大会影响延时时间的准确性，生产端实现复杂度大大增加。
使用批量消息，在一定程度上可以提高性能和吞吐量，但是确实也会存在一些问题，使用的时候要结合业务场景避开这些坑。
本文链接：http://www.28at.com/showinfo-26-35258-0.html消息队列批量收发消息，请避开这五个坑！
声明：本网页内容旨在传播知识，若有侵权等问题请及时与本网联系，我们将在第一时间删除处理。邮件：2376512515@qq.com
上一篇： 利用属性选择器对外部链接进行样式设计
Copyright © 2016-2023 天津谷骐科技有限公司 版权所有 sitemap.xml
违法及侵权请联系：2376512515@qq.com 津ICP备18001702号
津公网安备 12010102000574号