溪谷游戏工作室OA系统,贴合游戏公会实际运营情况,为游戏公会/工作室提供正规、合法的游戏工作室管理解决方案。能够从管理层面帮助游戏公会改变其散乱、无序的管理状态,实现数据互连、信息互通。可根据各推广员的能力、业绩情况设置不同的薪资待遇,实现人员管理及薪资体系一套系统统一管理,避免成员流失,为游戏公会/工作室管理者提升团队运作监管能力以及工作效率。
软件特点
1.精细化权限设置
自定义多级角色设置,实现字段级的权限设置、复杂的多层级部门管理和权限控制
2.数据整合
整合多平台数据,支持获取 多平台整合数据
3.成员管理
防范工作室人员流失,大大解决公司管理问题
4.游戏管理
打通多渠道游戏资源,游戏申请高效便捷
5.渠道对接
公会可自主对接渠道或直接激活渠道
6.安全防护
OA系统闭环管理,防范随意盗取公司客户资源问题
文件:590m.com/f/25127180-497690345-47eef0(访问密码:551685)
以下内容无关:
-------------------------------------------分割线---------------------------------------------
问题描述
查阅了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版本为
com.azure.spring azure-spring-cloud-stream-binder-eventhubs 2.4.0 在参考官网的示例后,使用Supplier方法发送消息,代替Send。经过多次测试,指定partitionkey 之后,发送消息是顺序发送的,消费的时候也是按照顺序消费的,下面是测试的代码和结果发送端的代码
复制代码
// Copyright © 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();
};
}
}