如何配置Azure Service Bus以批量发送消息？

Question

目的是消耗从Kafka并重定向到Azure Service Bus的记录。每次调查Kafka主题时，可以在调试时配置Kafka消费者并确认数千条消息。

另一方面，Azure Service Bus将始终发布一条消息。因此，当此集成运行时，Kafka日志将表明收到了数千条消息，然后Azure Service Bus Logs将迭代每次发送到队列的每个消息。在迭代中，这需要几分钟的时间，从而大大降低了过程。

组件文档批次是默认值，但在如何实现这一目标上还不清楚。

public class SampleKafkaConsumer extends RouteBuilder {
  @Override
  public void configure() throws Exception {
    log.info("About to start route: Kafka Server -> Log ");
    from("kafka:{{consumer.topic}}?brokers={{kafka.host}}:{{kafka.port}}"
             + "&maxPollRecords={{consumer.maxPollRecords}}"
             + "&consumersCount={{consumer.consumersCount}}"
             + "&seekTo={{consumer.seekTo}}"
             + "&groupId={{consumer.group}}"
             + "&lingerMs={{consumer.lingerMs}}"
             + "&producerBatchSize={{consumer.producerBatchSize}}"
             + "&saslJaasConfig={{consumer.saslJaasConfig}}"
             + "&saslMechanism={{consumer.saslMechanism}}"
             + "&securityProtocol={{consumer.securityProtocol}}")
             .routeId("Kafka")
             .to("azure-servicebus:topic?connectionString={{producer.connectionString}}&producerOperation=sendMessages");
             
  }
}

关于如何处理这个的见解吗？

Answer 1

Azure Service Bus文档试图暗示的正确解决方案是将结果作为对象或消息列表批量，并将其传递到Azure生产者中。

在apache骆驼中这样做的关键是使用。这是在下面的代码中具有完成间隔的夫妇，允许管道将消息串在一起并分批发送到服务总线。

public class SampleKafkaConsumer extends RouteBuilder {
  @Override
  public void configure() throws Exception {
    log.info("About to start route: Kafka Server -> Log ");
    from("kafka:{{consumer.topic}}?brokers={{kafka.host}}:{{kafka.port}}"
             + "&maxPollRecords={{consumer.maxPollRecords}}"
             + "&consumersCount={{consumer.consumersCount}}"
             + "&seekTo={{consumer.seekTo}}"
             + "&groupId={{consumer.group}}"
             + "&lingerMs={{consumer.lingerMs}}"
             + "&producerBatchSize={{consumer.producerBatchSize}}"
             + "&saslJaasConfig={{consumer.saslJaasConfig}}"
             + "&saslMechanism={{consumer.saslMechanism}}"
             + "&securityProtocol={{consumer.securityProtocol}}")
             .routeId("Kafka")
             .aggregate(new AzureAggregationStrategy()).constant(true)
             .completionInterval(300L)
             .to("azure-servicebus:topic?connectionString={{producer.connectionString}}&producerOperation=sendMessages");
             
  }
}