問(wèn)題描述
查閱了Azure的官方文檔( 將事件發(fā)送到特定分區(qū): 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”來(lái)連接EventHub發(fā)送和消費(fèi)消息事件枉证。在發(fā)送端一個(gè)For循環(huán)中發(fā)送帶順序號(hào)的消息癌椿，編號(hào)從0開(kāi)始，并且在消息的header中指定了 "Partition Key"忠售，相同PartitionKey的消息會(huì)被發(fā)送到相同的Partition，來(lái)保證這些消息的順序。
但是在消費(fèi)端的工程中消費(fèi)這些消息時(shí)，看到打印到日志中的結(jié)果并不是從0遞增的信夫。所以想知道是發(fā)送端在發(fā)送時(shí)就已經(jīng)亂序發(fā)送了？還是消息到達(dá)EventHub后亂序保存了？還是消費(fèi)端的消費(fèi)方式的問(wèn)題静稻，導(dǎo)致打印出的結(jié)果是亂序的警没？
下面是發(fā)送端的代碼：
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());
}
}
下面是消費(fèi)端代碼：
@StreamListener(Sink.INPUT)
public void onEvent(String message, @Header(AzureHeaders.CHECKPOINTER) Checkpointer checkpointer,
@Header(AzureHeaders.RAW_PARTITION_ID) String rawPartitionId,
@Header(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":""}
從日志中可以看到，消息確實(shí)都被發(fā)送到了同一個(gè)分區(qū)（rawPartitionId=1）振湾，但是從消息體的序號(hào)上看惠奸，是亂序的
問(wèn)題分析
這個(gè)是和這個(gè)配置相關(guān)的fixedDelay，指定默認(rèn)輪詢器的固定延遲恰梢，是一個(gè)周期性觸發(fā)器佛南，之前代碼會(huì)根據(jù)這個(gè)輪詢器進(jìn)行發(fā)送和接受消息的。使用Send發(fā)送的方法嵌言，現(xiàn)在最新的SDK 不使用這個(gè)方法嗅回，所以需要使用新的sdk 發(fā)送數(shù)據(jù)測(cè)試一下。
新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>
在參考官網(wǎng)的示例后摧茴，使用Supplier方法發(fā)送消息绵载，代替Send。經(jīng)過(guò)多次測(cè)試苛白，指定partitionkey 之后娃豹，發(fā)送消息是順序發(fā)送的，消費(fèi)的時(shí)候也是按照順序消費(fèi)的购裙，下面是測(cè)試的代碼和結(jié)果
發(fā)送端的代碼
// 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;
@Configuration
public class EventProducerConfiguration {
private static final Logger LOGGER = LoggerFactory.getLogger(EventProducerConfiguration.class);
private int i = 0;
@Bean
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;
@Configuration
public class EventConsume {
private static final Logger LOGGER = LoggerFactory.getLogger(EventConsume.class);
@Bean
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();
};
}
}
發(fā)送消息的日志
消費(fèi)消息的日志
參考資料
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
[END]
當(dāng)在復(fù)雜的環(huán)境中面臨問(wèn)題懂版，格物之道需：濁而靜之徐清，安以動(dòng)之徐生躏率。 云中躯畴，恰是如此!
分類: 【Azure 事件中心】
標(biāo)簽: 消費(fèi)端代碼, 事件指定順序, 事件中心 Azure Event Hub
