問題描述

查閱了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());

    }

}

下面是消費端程式碼：

@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":""}

從日誌中可以看到，訊息確實都被髮送到了同一個分割槽（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;

@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();

        };

    }

}

傳送訊息的日誌

消費訊息的日誌

參考資料

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]