问题描述
查阅了Azure的官方文档( 将事件发送到特定分区: https://docs.azure.cn/zh-cn/event-hubs/event-hubs-availability-and-consistency?tabs=java#send-events-to-a-specific-partition)，在工程里引用组件“azure-spring-cloud-stream-binder-eventhubs”来连接EventHub发送和消费消息事件。在发送端一个For循环中发送带顺序号的消息，编号从0开始，并且在消息的header中指定了 "Partition Key"，相同PartitionKey的消息会被发送到相同的Partition，来保证这些消息的顺序。
但是在消费端的工程中消费这些消息时，看到打印到日志中的结果并不是从0递增的。所以想知道是发送端在发送时就已经乱序发送了？还是消息到达EventHub后乱序保存了？还是消费端的消费方式的问题，导致打印出的结果是乱序的？
下面是发送端的代码：
public void testPushMessages(int mcount, String partitionKey) { String message = "Message "; for (int i=0; i <mcount; i++) { source.output().send(MessageBuilder.withPayload(partitionKey + mcount + i).setHeaderIfAbsent(AzureHeaders.PARTITION_KEY,partitionKey).build()); } }
下面是消费端代码：
(Sink.INPUT) public void onEvent(String message, (AzureHeaders.CHECKPOINTER) Checkpointer checkpointer, (AzureHeaders.RAW_PARTITION_ID) String rawPartitionId, (AzureHeaders.PARTITION_KEY) String partitionKey) { checkpointer.success() .doOnSuccess(s -> log.info("Message '{}' successfully check pointed.rawPartitionId={},partitionKey={}", message, rawPartitionId, partitionKey)) .doOnError(s -> log.error("Checkpoint message got exception.")) .subscribe();
下面是打印的日志
......,"data":"Message 'testKey4testMessage1' successfully check pointed.rawPartitionId=1,partition<*****>","xcptn":""} ......,"data":"Message 'testKey5testMessage29' successfully check pointed.rawPartitionId=1,partition<*****>","xcptn":""} ......,"data":"Message 'testKey5testMessage27' successfully check pointed.rawPartitionId=1,partition<*****>","xcptn":""} ......,"data":"Message 'testKey5testMessage26' successfully check pointed.rawPartitionId=1,partition<*****>","xcptn":""} ......,"data":"Message 'testKey5testMessage25' successfully check pointed.rawPartitionId=1,partition<*****>","xcptn":""} ......,"data":"Message 'testKey5testMessage28' successfully check pointed.rawPartitionId=1,partition<*****>","xcptn":""} ......,"data":"Message 'testKey5testMessage14' successfully check pointed.rawPartitionId=1,partition<*****>","xcptn":""} ......,"data":"Message 'testKey5testMessage13' successfully check pointed.rawPartitionId=1,partition<*****>","xcptn":""} ......,"data":"Message 'testKey5testMessage15' successfully check pointed.rawPartitionId=1,partition<*****>","xcptn":""} ......,"data":"Message 'testKey5testMessage5' successfully check pointed.rawPartitionId=1,partition<*****>","xcptn":""} ......,"data":"Message 'testKey5testMessage7' successfully check pointed.rawPartitionId=1,partition<*****>","xcptn":""} ......,"data":"Message 'testKey5testMessage20' successfully check pointed.rawPartitionId=1,partition<*****>","xcptn":""} ......,"data":"Message 'testKey5testMessage19' successfully check pointed.rawPartitionId=1,partition<*****>","xcptn":""} ......,"data":"Message 'testKey5testMessage18' successfully check pointed.rawPartitionId=1,partition<*****>","xcptn":""} ......,"data":"Message 'testKey5testMessage0' successfully check pointed.rawPartitionId=1,partition<*****>","xcptn":""} ......,"data":"Message 'testKey5testMessage9' successfully check pointed.rawPartitionId=1,partition<*****>","xcptn":""} ......,"data":"Message 'testKey5testMessage12' successfully check pointed.rawPartitionId=1,partition<*****>","xcptn":""} ......,"data":"Message 'testKey5testMessage8' successfully check pointed.rawPartitionId=1,partition<*****>","xcptn":""}
从日志中可以看到，消息确实都被发送到了同一个分区（rawPartitionId=1），但是从消息体的序号上看，是乱序的
问题分析
这个是和这个配置相关的fixedDelay，指定默认轮询器的固定延迟，是一个周期性触发器，之前代码会根据这个轮询器进行发送和接受消息的。使用Send发送的方法，现在最新的SDK 不使用这个方法，所以需要使用新的sdk 发送数据测试一下。
新sdk 参考文档您可以参考一下：https://github.com/Azure/azure-sdk-for-java/tree/master/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-binder
SDK版本为
<dependency>
<groupId>com.azure.spring</groupId>
<artifactId>azure-spring-cloud-stream-binder-eventhubs</artifactId>
<version>2.4.0</version>
</dependency>
在参考官网的示例后，使用Supplier方法发送消息，代替Send。经过多次测试，指定partitionkey 之后，发送消息是顺序发送的，消费的时候也是按照顺序消费的，下面是测试的代码和结果
发送端的代码
// Copyright (c) Microsoft Corporation. All rights reserved. // Licensed under the MIT License. package com.azure.spring.sample.eventhubs.binder; import com.azure.spring.integration.core.EventHubHeaders; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Profile; import org.springframework.messaging.Message; import org.springframework.messaging.support.MessageBuilder; import java.util.function.Supplier; import static com.azure.spring.integration.core.EventHubHeaders.SEQUENCE_NUMBER; public class EventProducerConfiguration { private static final Logger LOGGER = LoggerFactory.getLogger(EventProducerConfiguration.class); private int i = 0; public Supplier<Message<String>> supply() { return () -> { //LOGGER.info("Sending message, sequence " + i); String partitionKey="info"; LOGGER.info("Send message " + MessageBuilder.withPayload("hello world, "+i).setHeaderIfAbsent(EventHubHeaders.PARTITION_KEY, partitionKey).build()); return MessageBuilder.withPayload("hello world, "+ i++). setHeaderIfAbsent(EventHubHeaders.PARTITION_KEY, partitionKey).build(); }; } }
接收端的代码
package com.ywt.demoEventhub; import com.azure.spring.integration.core.EventHubHeaders; import com.azure.spring.integration.core.api.reactor.Checkpointer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.integration.annotation.ServiceActivator; import org.springframework.messaging.Message; import java.util.function.Consumer; import static com.azure.spring.integration.core.AzureHeaders.CHECKPOINTER; public class EventConsume { private static final Logger LOGGER = LoggerFactory.getLogger(EventConsume.class); public Consumer<Message<String>> consume() { return message -> { Checkpointer checkpointer = (Checkpointer) message.getHeaders().get(CHECKPOINTER); LOGGER.info("New message received: '{}', partition key: {}, sequence number: {}, offset: {}, enqueued time: {}", message.getPayload(), message.getHeaders().get(EventHubHeaders.PARTITION_KEY), message.getHeaders().get(EventHubHeaders.SEQUENCE_NUMBER), message.getHeaders().get(EventHubHeaders.OFFSET), message.getHeaders().get(EventHubHeaders.ENQUEUED_TIME) ); checkpointer.success() .doOnSuccess(success -> LOGGER.info("Message '{}' successfully checkpointed number '{}' ", message.getPayload(), message.getHeaders().get(EventHubHeaders.CHECKPOINTER))) .doOnError(error -> LOGGER.error("Exception found", error)) .subscribe(); }; } }
发送消息的日志
消费消息的日志
参考资料
Azure Spring Cloud Stream Binder for Event Hub Code Sample shared library for Java：https://github.com/Azure/azure-sdk-for-java/tree/master/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-binder
How to create a Spring Cloud Stream Binder application with Azure Event Hubs - Add sample code to implement basic event hub functionality : https://docs.microsoft.com/en-us/azure/developer/java/spring-framework/configure-spring-cloud-stream-binder-java-app-azure-event-hub#add-sample-code-to-implement-basic-event-hub-functionality
