如何配置Azure Service Bus实现消息批量发送

当前配置没有触发Azure Service Bus（以下简称ASB）批量发送，核心原因有两个：

1. Camel Kafka消费者默认将单次poll拉取到的批量消息拆分为独立的单个Exchange逐条向下游传递，不会把整批消息作为一个集合交给ASB组件。你观察到的Kafka端拉取数千条消息后ASB逐条发送，就是这个默认行为导致的。
2. Camel ASB组件的sendMessages操作不会自动聚合零散消息，只有当传入的Exchange消息体是Iterable<ServiceBusMessage>类型的消息集合时，才会调用ASB的批量发送API；传入单条消息时会自动退化为单条发送，和文档提到的“批量发送默认启用”并不矛盾——组件默认支持批量能力，但不会主动攒批。

• 开启Kafka消费者的批处理模式
  在Kafka端点的连接参数中追加&batching=true配置。该参数默认值为false，开启后Camel不会拆分单次poll拉取的消息，而是将整批消息封装为List类型的消息体，一次性传递给下游路由，避免逐条Exchange流转的开销。

  补充说明：你原有Kafka消费者配置里的lingerMs、producerBatchSize是Kafka生产者的专属参数，放在消费者端点上不会生效，可以直接移除，避免无效配置干扰排查。

• 做消息类型转换适配ASB批量接口
  Kafka传递过来的List是原始消息值的集合，需要转换为ASB要求的List<ServiceBusMessage>实例，才能被批量接口识别。
• 对齐ASB批量限制配置参数
  ASB服务端对单批请求有硬限制：单批最多包含100条消息、总载荷不超过256KB。你可以根据消息大小在ASB端点配置maxBatchSize、maxBatchSizeInBytes参数，避免请求被服务端拒绝；如果Kafka端maxPollRecords设置超过100，需要在转换逻辑中手动拆分批次，不要一次性把超过100条的集合传给ASB端点。

public class SampleKafkaConsumer extends RouteBuilder {
  @Override
  public void configure() throws Exception {
    log.info("About to start route: Kafka Server -> Azure Service Bus");
    from("kafka:{{consumer.topic}}?brokers={{kafka.host}}:{{kafka.port}}"
             + "&maxPollRecords={{consumer.maxPollRecords}}" // 建议设置为≤100，匹配ASB单批上限
             + "&batching=true" // 核心：开启Kafka消费者批处理，不拆分单条Exchange
             + "&consumersCount={{consumer.consumersCount}}"
             + "&seekTo={{consumer.seekTo}}"
             + "&groupId={{consumer.group}}"
             + "&saslJaasConfig={{consumer.saslJaasConfig}}"
             + "&saslMechanism={{consumer.saslMechanism}}"
             + "&securityProtocol={{consumer.securityProtocol}}")
             .routeId("KafkaToASB")
             // 转换Kafka原始消息集合为ASB要求的ServiceBusMessage集合
             .process(exchange -> {
                 List<String> kafkaMessages = exchange.getIn().getBody(List.class);
                 List<ServiceBusMessage> asbBatchMessages = kafkaMessages.stream()
                         .map(ServiceBusMessage::new)
                         .collect(Collectors.toList());
                 exchange.getIn().setBody(asbBatchMessages);
             })
             .to("azure-servicebus:topic?connectionString={{producer.connectionString}}"
             + "&producerOperation=sendMessages"
             + "&maxBatchSize=100" // 对齐ASB单批最大消息数限制
             + "&maxBatchSizeInBytes=262144"); // 对齐ASB单批最大256KB载荷限制
  }
}

注意：代码中使用的ServiceBusMessage类为com.azure.messaging.servicebus.ServiceBusMessage，导入时注意包路径不要引错。

配置生效后，可以通过Camel的debug日志观察到ASB组件打印批量发送相关日志，不再出现逐条发送的记录；单次poll千条级别的消息转发耗时会从数分钟降低到秒级，和单条发送相比吞吐量提升10~100倍不等。

内容的提问来源于stack exchange，提问作者kvas1r