Azure事件中心消息乱序

摘要

Azure的手机客户端部件遇到了难题,实践活动信息无法按照次序达到。官方文档提供了解决方案,可以将事情发送至特殊系统分区。

正文

【Azure 事情管理中心】azure-spring-cloud-stream-binder-eventhubs手机客户端部件难题, 实践活动信息非次序可以达到

难题叙述

查看了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]

 

关注不迷路

扫码下方二维码,关注宇凡盒子公众号,免费获取最新技术内幕!

温馨提示:如果您访问和下载本站资源,表示您已同意只将下载文件用于研究、学习而非其他用途。
文章版权声明 1、本网站名称:宇凡盒子
2、本站文章未经许可,禁止转载!
3、如果文章内容介绍中无特别注明,本网站压缩包解压需要密码统一是:yufanbox.com
4、本站仅供资源信息交流学习,不保证资源的可用及完整性,不提供安装使用及技术服务。点此了解
5、如果您发现本站分享的资源侵犯了您的权益,请及时通知我们,我们会在接到通知后及时处理!提交入口
0

评论0

请先

站点公告

🚀 【宇凡盒子】全网资源库转储中心

👉 注册即送VIP权限👈

👻 全站资源免费下载✅,欢迎注册!

记得 【收藏】+【关注】 谢谢!~~~

立即注册
没有账号?注册  忘记密码?

社交账号快速登录