the expected type is FIFO, but actual type is NORMAL BROKER #785
-
|
为何我使用rocketmq-spring `发送消息时，总是提示实际broker为normal，请问是少了什么配置吗？
创建主题 发送方法 报错 |
Beta Was this translation helpful? Give feedback.
Replies: 2 comments 18 replies
-
|
发送顺序消息时需要的是 FIFO 类型的 topic，但实际上这个 topic 是一个 Normal 类型的。 |
Beta Was this translation helpful? Give feedback.
-
|
有部署控制台，不过在哪里看呢？没看到有这个相关的属性，另外请问我创建topic的命令有问题吗？
|
Beta Was this translation helpful? Give feedback.
-
|
查看方式：在控制台的 tipic 页面下，选择 FIFO 类型的消息进行过滤，就可以看到对应的 topic 了。 但在你的第一张截图左上角里面，我看到你选中了普通消息，不确定是否也勾选了 FIFO 类型，这个你自己可以确认一下，如果没有，则说明这个 topic 不是 FIFO 类型的。 关于你的这个问题，从你的异常日志里面输出的异常堆栈信息，可以看到你使用的是 remoting 协议的 sdk，同时，你的接入点（也就是配置里面的 name server 地址）应该是 proxy 地址。 我大概验证了一下，确实跟你出现了相同的错误，排查后发现是 proxy 当中这部分代码： protected RemotingCommand sendMessage(ChannelHandlerContext ctx, RemotingCommand request,
ProxyContext context) throws Exception {
SendMessageRequestHeader requestHeader = SendMessageRequestHeader.parseRequestHeader(request);
String topic = requestHeader.getTopic();
Map<String, String> property = MessageDecoder.string2messageProperties(requestHeader.getProperties());
// 这里获取到消息的类型
TopicMessageType messageType = TopicMessageType.parseFromMessageProperty(property);
if (ConfigurationManager.getProxyConfig().isEnableTopicMessageTypeCheck()) {
if (topicMessageTypeValidator != null) {
// Do not check retry or dlq topic
if (!NamespaceUtil.isRetryTopic(topic) && !NamespaceUtil.isDLQTopic(topic)) {
TopicMessageType topicMessageType = messagingProcessor.getMetadataService().getTopicMessageType(context, topic);
// 这一步校验出错了
topicMessageTypeValidator.validate(topicMessageType, messageType);
}
}
}
......
}其中获取消息类型的代码： public static TopicMessageType parseFromMessageProperty(Map<String, String> messageProperty) {
String isTrans = messageProperty.get(MessageConst.PROPERTY_TRANSACTION_PREPARED);
String isTransValue = "true";
if (isTransValue.equals(isTrans)) {
return TopicMessageType.TRANSACTION;
} else if (messageProperty.get(MessageConst.PROPERTY_DELAY_TIME_LEVEL) != null
|| messageProperty.get(MessageConst.PROPERTY_TIMER_DELIVER_MS) != null
|| messageProperty.get(MessageConst.PROPERTY_TIMER_DELAY_SEC) != null
|| messageProperty.get(MessageConst.PROPERTY_TIMER_DELAY_MS) != null) {
return TopicMessageType.DELAY;
} else if (messageProperty.get(MessageConst.PROPERTY_SHARDING_KEY) != null) {
// 通过判断是有这个属性来确定是 FIFO 类型
return TopicMessageType.FIFO;
}
return TopicMessageType.NORMAL;
}大概整理一下思路： 我不确定是不是个 bug，这个需要官方的人来回答了。我这边有临时的解决方案：
msg.putUserProperty("__SHARDINGKEY", 分组 key);
|
Beta Was this translation helpful? Give feedback.
-
|
我使用的是rocketmq spring进行发送的消息 |
Beta Was this translation helpful? Give feedback.
-
|
我按照您的说法增加了 __SHARDINGKEY 确实可以收发消息了，但是这个分布式键属性的意义和作用是什么呢 |
Beta Was this translation helpful? Give feedback.
-
我也是这个源码编译出来的，你试试最右侧切换版本，切成 5.x 的版本 |
Beta Was this translation helpful? Give feedback.
-
为了保证同一个 ShardingKey 的消息发送到相同队列上
新出了一个版本 rocketmq-v5-client-spring-boot ，内部使用的 grpc sdk。 |
Beta Was this translation helpful? Give feedback.
-
|
切换后可以显示了，另外请教下您那边的消费者终端能正常显示吗？我点击所有终端都会报错 |
Beta Was this translation helpful? Give feedback.
-
|
这个应该是 bug，看了下代码，这个终端的查询逻辑如下： // 获取 group 信息
ConsumerGroupInfo consumerGroupInfo =
this.brokerController.getConsumerManager().getConsumerGroupInfo(requestHeader.getConsumerGroup(), false);
if (consumerGroupInfo != null) {
// .... 省略构建参数
// 获取 channel 信息，构建城客户端信息
Iterator<Map.Entry<Channel, ClientChannelInfo>> it = consumerGroupInfo.getChannelInfoTable().entrySet().iterator();
while (it.hasNext()) {
ClientChannelInfo info = it.next().getValue();
Connection connection = new Connection();
connection.setClientId(info.getClientId());
connection.setLanguage(info.getLanguage());
connection.setVersion(info.getVersion());
connection.setClientAddr(RemotingHelper.parseChannelRemoteAddr(info.getChannel()));
bodydata.getConnectionSet().add(connection);
}
}
public ConsumerGroupInfo getConsumerGroupInfo(String group, boolean fromCompensationTable) {
// consumerTable 里面没有 group 信息
ConsumerGroupInfo consumerGroupInfo = consumerTable.get(group);
// fromCompensationTable 传的 false 所以也不会从补偿表里面获取
if (consumerGroupInfo == null && fromCompensationTable) {
consumerGroupInfo = consumerCompensationTable.get(group);
}
return consumerGroupInfo;
}主要原因是如果 client 是和 proxy 建连的话，订阅信息是在拉取消息的时候做补偿设置的，所以这部分数据存放在 |
Beta Was this translation helpful? Give feedback.
-
除了其他控制台的问题，能正常消费了，非常感谢 |
Beta Was this translation helpful? Give feedback.
-
控制台的那个问题我也找到了，提了个 pr，你可以参考一下自己改掉后打包。 |
Beta Was this translation helpful? Give feedback.
为了保证同一个 ShardingKey 的消息发送到相同队列上
新出了一个版本 rocketmq-v5-client-spring-boot ，内部使用的 grpc sdk。