You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@rocketmq.apache.org by "loringgit (via GitHub)" <gi...@apache.org> on 2025/04/01 02:51:21 UTC
[D] FIFO topic 消费端非严格有序 [rocketmq]
GitHub user loringgit created a discussion: FIFO topic 消费端非严格有序
背景：
1.rockermq 5.3 版本，客户端版本：rocketmq-clients/golang/v5
2.创建FIFO topic，2 个queue
3.Consumer group 下有两个consumer实例
4.producer 发送消息的message group 都是fifo001，也就是只有一个queue有消息
'''
Queue | minOffset | maxOffset | lastUpdateTimeStamp
-- | -- | -- | --
MessageQueue [topic=test_fifo_topic, brokerName=cf8605034927, queueId=1] | 0 | 0 | 1970-01-01 08:00:00
MessageQueue [topic=test_fifo_topic, brokerName=cf8605034927, queueId=0] | 0 | 1000 | 2025-04-01 09:28:13
'''
5.consumer 端代码参考的是example中SimpleConsumer，流程：收到消息后打印log=>模拟消息处理sleep一秒=> 打印log=> ACK，具体代码如下：
'''
func main() {
// log to console
os.Setenv("mq.consoleAppender.enabled", "true")
rmq_client.ResetLogger()
// In most case, you don't need to create many consumers, singleton pattern is more recommended.
simpleConsumer, err := rmq_client.NewSimpleConsumer(&rmq_client.Config{
Endpoint: Endpoint,
ConsumerGroup: ConsumerGroup,
Credentials: &credentials.SessionCredentials{
AccessKey: AccessKey,
AccessSecret: SecretKey,
},
},
rmq_client.WithAwaitDuration(awaitDuration),
rmq_client.WithSubscriptionExpressions(map[string]*rmq_client.FilterExpression{
Topic: rmq_client.SUB_ALL,
}),
)
if err != nil {
log.Fatal(err)
}
// start simpleConsumer
err = simpleConsumer.Start()
if err != nil {
log.Fatal(err)
}
// graceful stop simpleConsumer
defer simpleConsumer.GracefulStop()
go func() {
fmt.Println("start receive message")
for {
mvs, err := simpleConsumer.Receive(context.TODO(), maxMessageNum, invisibleDuration)
if err != nil {
fmt.Println(err)
}
// ack message
for _, mv := range mvs {
fmt.Printf("message receive group:%s, timestamp:%d, offset:%d,body:%s\n",
*mv.GetMessageGroup(), time.Now().UnixMilli(), mv.GetOffset(), mv.GetBody())
if strings.EqualFold(*mv.GetMessageGroup(), "fifo001") {
time.Sleep(time.Second * 1)
}
fmt.Printf("message receive group:%s, timestamp:%d, offset:%d,body:%s,key=%v TO_ACK\n",
*mv.GetMessageGroup(), time.Now().UnixMilli(), mv.GetOffset(), mv.GetBody(), mv.GetKeys())
err = simpleConsumer.Ack(context.TODO(), mv)
if err != nil {
fmt.Println("simpleConsumer.Ack, err:", err)
}
}
fmt.Println()
}
}()
'''
测试打印log如下：
'''
consumer 1 log如下：
selectMessageQueueId: 1
message receive group:fifo001, timestamp:1743471018946, offset:5,body:message_5
message receive group:fifo001, timestamp:1743471019947, offset:5,body:message_5,key=[5] TO_ACK
selectMessageQueueId: 0
message receive group:fifo001, timestamp:1743471019984, offset:7,body:message_7
message receive group:fifo001, timestamp:1743471020985, offset:7,body:message_7,key=[7] TO_ACK
consumer2 log如下：
selectMessageQueueId: 0
message receive group:fifo001, timestamp:1743471019768, offset:6,body:message_6,key=[6]
message receive group:fifo001, timestamp:1743471020793, offset:6,body:message_6,key=[6] TO_ACK
selectMessageQueueId: 1
message receive group:fifo001, timestamp:1743471020806, offset:8,body:message_8,key=[8]
message receive group:fifo001, timestamp:1743471021819, offset:8,body:message_8,key=[8] TO_ACK
''''
问题1 ：顺序消息顺序性
从文档看，[顺序消息负载均衡文档说明](https://rocketmq.apache.org/zh/docs/featureBehavior/08consumerloadbalance/)，“在消费过程中，前面的消息M1、M2被消费者Consumer A1处理时，只要消费状态没有提交，消费者A2是无法并行消费后续的M3、M4消息的，必须等前面的消息提交消费状态后才能消费后面的消息”。但是从日志看，M3，M4 在M2 未ACK前已经收到。这是由broker端处理由问题，还是我的用法不对？
consumer 2收到消息：timestamp:1743471019768, offset:6
consumer 1 收到消息：timestamp:1743471018946, offset:5，
准备ack： timestamp:1743471019947, offset:5,body:message_5,key=[5] TO_ACK
从这里看到consumer1 的offset5消息尚未ack，consumer2 已经收到了offset 6 的消息
问题2 ：queue 0 没有消息，client 轮询选择了queue0的时，也有消息返回，broker是怎么处理的？
GitHub link: https://github.com/apache/rocketmq/discussions/9306
----
This is an automatically sent email for dev@rocketmq.apache.org.
To unsubscribe, please send an email to: dev-unsubscribe@rocketmq.apache.org
Re: [D] FIFO topic 消费端非严格有序 [rocketmq]
Posted by "francisoliverlee (via GitHub)" <gi...@apache.org>.
GitHub user francisoliverlee added a comment to the discussion: FIFO topic 消费端非严格有序
现在golang应该不支持fifo，grpc协议中的fifo没用被使用。
GitHub link: https://github.com/apache/rocketmq/discussions/9306#discussioncomment-12707993
----
This is an automatically sent email for dev@rocketmq.apache.org.
To unsubscribe, please send an email to: dev-unsubscribe@rocketmq.apache.org
Re: [D] FIFO topic 消费端非严格有序 [rocketmq]
Posted by "loringgit (via GitHub)" <gi...@apache.org>.
GitHub user loringgit edited a discussion: FIFO topic 消费端非严格有序
背景：
1.rockermq 5.3 版本，客户端版本：rocketmq-clients/golang/v5
2.创建FIFO topic，2 个queue
3.Consumer group 下有两个consumer实例
4.producer 发送消息的message group 都是fifo001，也就是只有一个queue有消息
'''
Queue | minOffset | maxOffset | lastUpdateTimeStamp
-- | -- | -- | --
MessageQueue [topic=test_fifo_topic, brokerName=cf8605034927, queueId=1] | 0 | 0 | 1970-01-01 08:00:00
MessageQueue [topic=test_fifo_topic, brokerName=cf8605034927, queueId=0] | 0 | 1000 | 2025-04-01 09:28:13
'''
5.consumer 端代码参考的是example中SimpleConsumer，流程：收到消息后打印log=>模拟消息处理sleep一秒=> 打印log=> ACK，具体代码如下：
```
func main() {
// log to console
os.Setenv("mq.consoleAppender.enabled", "true")
rmq_client.ResetLogger()
// In most case, you don't need to create many consumers, singleton pattern is more recommended.
simpleConsumer, err := rmq_client.NewSimpleConsumer(&rmq_client.Config{
Endpoint: Endpoint,
ConsumerGroup: ConsumerGroup,
Credentials: &credentials.SessionCredentials{
AccessKey: AccessKey,
AccessSecret: SecretKey,
},
},
rmq_client.WithAwaitDuration(awaitDuration),
rmq_client.WithSubscriptionExpressions(map[string]*rmq_client.FilterExpression{
Topic: rmq_client.SUB_ALL,
}),
)
if err != nil {
log.Fatal(err)
}
// start simpleConsumer
err = simpleConsumer.Start()
if err != nil {
log.Fatal(err)
}
// graceful stop simpleConsumer
defer simpleConsumer.GracefulStop()
go func() {
fmt.Println("start receive message")
for {
mvs, err := simpleConsumer.Receive(context.TODO(), maxMessageNum, invisibleDuration)
if err != nil {
fmt.Println(err)
}
// ack message
for _, mv := range mvs {
fmt.Printf("message receive group:%s, timestamp:%d, offset:%d,body:%s\n",
*mv.GetMessageGroup(), time.Now().UnixMilli(), mv.GetOffset(), mv.GetBody())
if strings.EqualFold(*mv.GetMessageGroup(), "fifo001") {
time.Sleep(time.Second * 1)
}
fmt.Printf("message receive group:%s, timestamp:%d, offset:%d,body:%s,key=%v TO_ACK\n",
*mv.GetMessageGroup(), time.Now().UnixMilli(), mv.GetOffset(), mv.GetBody(), mv.GetKeys())
err = simpleConsumer.Ack(context.TODO(), mv)
if err != nil {
fmt.Println("simpleConsumer.Ack, err:", err)
}
}
fmt.Println()
}
}()
```
测试打印log如下：
```
consumer 1 log如下：
selectMessageQueueId: 1
message receive group:fifo001, timestamp:1743471018946, offset:5,body:message_5
message receive group:fifo001, timestamp:1743471019947, offset:5,body:message_5,key=[5] TO_ACK
selectMessageQueueId: 0
message receive group:fifo001, timestamp:1743471019984, offset:7,body:message_7
message receive group:fifo001, timestamp:1743471020985, offset:7,body:message_7,key=[7] TO_ACK
consumer2 log如下：
selectMessageQueueId: 0
message receive group:fifo001, timestamp:1743471019768, offset:6,body:message_6,key=[6]
message receive group:fifo001, timestamp:1743471020793, offset:6,body:message_6,key=[6] TO_ACK
selectMessageQueueId: 1
message receive group:fifo001, timestamp:1743471020806, offset:8,body:message_8,key=[8]
message receive group:fifo001, timestamp:1743471021819, offset:8,body:message_8,key=[8] TO_ACK
```
问题1 ：顺序消息顺序性
从文档看，[顺序消息负载均衡文档说明](https://rocketmq.apache.org/zh/docs/featureBehavior/08consumerloadbalance/)，“在消费过程中，前面的消息M1、M2被消费者Consumer A1处理时，只要消费状态没有提交，消费者A2是无法并行消费后续的M3、M4消息的，必须等前面的消息提交消费状态后才能消费后面的消息”。但是从日志看，M3，M4 在M2 未ACK前已经收到。这是由broker端处理由问题，还是我的用法不对？
consumer 2收到消息：timestamp:1743471019768, offset:6
consumer 1 收到消息：timestamp:1743471018946, offset:5，
准备ack： timestamp:1743471019947, offset:5,body:message_5,key=[5] TO_ACK
从这里看到consumer1 的offset5消息尚未ack，consumer2 已经收到了offset 6 的消息
问题2 ：queue 0 没有消息，client 轮询选择了queue0的时，也有消息返回，broker是怎么处理的？
GitHub link: https://github.com/apache/rocketmq/discussions/9306
----
This is an automatically sent email for dev@rocketmq.apache.org.
To unsubscribe, please send an email to: dev-unsubscribe@rocketmq.apache.org
Re: [D] FIFO topic 消费端非严格有序 [rocketmq]
Posted by "loringgit (via GitHub)" <gi...@apache.org>.
GitHub user loringgit edited a discussion: FIFO topic 消费端非严格有序
背景：
1.rockermq 5.3 版本，客户端版本：rocketmq-clients/golang/v5
2.创建FIFO topic，2 个queue
3.Consumer group 下有两个consumer实例
4.producer 发送消息的message group 都是fifo001，也就是只有一个queue有消息
Queue | minOffset | maxOffset | lastUpdateTimeStamp
-- | -- | -- | --
MessageQueue [topic=test_fifo_topic, brokerName=cf8605034927, queueId=1] | 0 | 0 | 1970-01-01 08:00:00
MessageQueue [topic=test_fifo_topic, brokerName=cf8605034927, queueId=0] | 0 | 1000 | 2025-04-01 09:28:13
5.consumer 端代码参考的是example中SimpleConsumer，流程：收到消息后打印log=>模拟消息处理sleep一秒=> 打印log=> ACK，具体代码如下：
```
func main() {
// log to console
os.Setenv("mq.consoleAppender.enabled", "true")
rmq_client.ResetLogger()
// In most case, you don't need to create many consumers, singleton pattern is more recommended.
simpleConsumer, err := rmq_client.NewSimpleConsumer(&rmq_client.Config{
Endpoint: Endpoint,
ConsumerGroup: ConsumerGroup,
Credentials: &credentials.SessionCredentials{
AccessKey: AccessKey,
AccessSecret: SecretKey,
},
},
rmq_client.WithAwaitDuration(awaitDuration),
rmq_client.WithSubscriptionExpressions(map[string]*rmq_client.FilterExpression{
Topic: rmq_client.SUB_ALL,
}),
)
if err != nil {
log.Fatal(err)
}
// start simpleConsumer
err = simpleConsumer.Start()
if err != nil {
log.Fatal(err)
}
// graceful stop simpleConsumer
defer simpleConsumer.GracefulStop()
go func() {
fmt.Println("start receive message")
for {
mvs, err := simpleConsumer.Receive(context.TODO(), maxMessageNum, invisibleDuration)
if err != nil {
fmt.Println(err)
}
// ack message
for _, mv := range mvs {
fmt.Printf("message receive group:%s, timestamp:%d, offset:%d,body:%s\n",
*mv.GetMessageGroup(), time.Now().UnixMilli(), mv.GetOffset(), mv.GetBody())
if strings.EqualFold(*mv.GetMessageGroup(), "fifo001") {
time.Sleep(time.Second * 1)
}
fmt.Printf("message receive group:%s, timestamp:%d, offset:%d,body:%s,key=%v TO_ACK\n",
*mv.GetMessageGroup(), time.Now().UnixMilli(), mv.GetOffset(), mv.GetBody(), mv.GetKeys())
err = simpleConsumer.Ack(context.TODO(), mv)
if err != nil {
fmt.Println("simpleConsumer.Ack, err:", err)
}
}
fmt.Println()
}
}()
```
测试打印log如下：
```
consumer 1 log如下：
selectMessageQueueId: 1
message receive group:fifo001, timestamp:1743471018946, offset:5,body:message_5
message receive group:fifo001, timestamp:1743471019947, offset:5,body:message_5,key=[5] TO_ACK
selectMessageQueueId: 0
message receive group:fifo001, timestamp:1743471019984, offset:7,body:message_7
message receive group:fifo001, timestamp:1743471020985, offset:7,body:message_7,key=[7] TO_ACK
consumer2 log如下：
selectMessageQueueId: 0
message receive group:fifo001, timestamp:1743471019768, offset:6,body:message_6,key=[6]
message receive group:fifo001, timestamp:1743471020793, offset:6,body:message_6,key=[6] TO_ACK
selectMessageQueueId: 1
message receive group:fifo001, timestamp:1743471020806, offset:8,body:message_8,key=[8]
message receive group:fifo001, timestamp:1743471021819, offset:8,body:message_8,key=[8] TO_ACK
```
问题1 ：顺序消息顺序性
从文档看，[顺序消息负载均衡文档说明](https://rocketmq.apache.org/zh/docs/featureBehavior/08consumerloadbalance/)，“在消费过程中，前面的消息M1、M2被消费者Consumer A1处理时，只要消费状态没有提交，消费者A2是无法并行消费后续的M3、M4消息的，必须等前面的消息提交消费状态后才能消费后面的消息”。但是从日志看，M3，M4 在M2 未ACK前已经收到。这是由broker端处理由问题，还是我的用法不对？
```
consumer 2收到消息：timestamp:1743471019768, offset:6
consumer 1 收到消息：timestamp:1743471018946, offset:5，
consumer 1 准备ack 时刻： timestamp:1743471019947, offset:5,body:message_5,key=[5] TO_ACK
```
从这里看到consumer1 的offset5消息尚未ack，consumer2 已经收到了offset 6 的消息
问题2 ：queue 0 没有消息，client 轮询选择了queue0的时，也有消息返回，broker是怎么处理的？
GitHub link: https://github.com/apache/rocketmq/discussions/9306
----
This is an automatically sent email for dev@rocketmq.apache.org.
To unsubscribe, please send an email to: dev-unsubscribe@rocketmq.apache.org
Re: [D] FIFO topic 消费端非严格有序 [rocketmq]
Posted by "guyinyou (via GitHub)" <gi...@apache.org>.
GitHub user guyinyou added a comment to the discussion: FIFO topic 消费端非严格有序
目前有一版 go pushconsumer的实现，里面有fifo的支持，感兴趣的可以一起review一下：
https://github.com/apache/rocketmq-clients/pull/990
GitHub link: https://github.com/apache/rocketmq/discussions/9306#discussioncomment-13187907
----
This is an automatically sent email for dev@rocketmq.apache.org.
To unsubscribe, please send an email to: dev-unsubscribe@rocketmq.apache.org
Re: [D] FIFO topic 消费端非严格有序 [rocketmq]
Posted by "francisoliverlee (via GitHub)" <gi...@apache.org>.
GitHub user francisoliverlee added a comment to the discussion: FIFO topic 消费端非严格有序
问题1
创建topic的时候，需要选择topic类型为fifo。 消费fifo java push consumer中存在，
https://github.com/apache/rocketmq-clients/blob/109dbc40b1ada77f2de56c453994259f86e8ae88/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/PushSubscriptionSettings.java#L49
golang中也对应找一下呢
问题2
日志和代码没有对上。 日志中存在selectMessageQueueId关键， 代码没有。 可以贴下实际的代码
GitHub link: https://github.com/apache/rocketmq/discussions/9306#discussioncomment-12694888
----
This is an automatically sent email for dev@rocketmq.apache.org.
To unsubscribe, please send an email to: dev-unsubscribe@rocketmq.apache.org
Re: [D] FIFO topic 消费端非严格有序 [rocketmq]
Posted by "loringgit (via GitHub)" <gi...@apache.org>.
GitHub user loringgit edited a discussion: FIFO topic 消费端非严格有序
背景：
1.rockermq 5.3 版本，客户端版本：rocketmq-clients/golang/v5
2.创建FIFO topic，2 个queue
3.Consumer group 下有两个consumer实例
4.producer 发送消息的message group 都是fifo001，也就是只有一个queue有消息
'''
Queue | minOffset | maxOffset | lastUpdateTimeStamp
-- | -- | -- | --
MessageQueue [topic=test_fifo_topic, brokerName=cf8605034927, queueId=1] | 0 | 0 | 1970-01-01 08:00:00
MessageQueue [topic=test_fifo_topic, brokerName=cf8605034927, queueId=0] | 0 | 1000 | 2025-04-01 09:28:13
'''
5.consumer 端代码参考的是example中SimpleConsumer，流程：收到消息后打印log=>模拟消息处理sleep一秒=> 打印log=> ACK，具体代码如下：
```
func main() {
// log to console
os.Setenv("mq.consoleAppender.enabled", "true")
rmq_client.ResetLogger()
// In most case, you don't need to create many consumers, singleton pattern is more recommended.
simpleConsumer, err := rmq_client.NewSimpleConsumer(&rmq_client.Config{
Endpoint: Endpoint,
ConsumerGroup: ConsumerGroup,
Credentials: &credentials.SessionCredentials{
AccessKey: AccessKey,
AccessSecret: SecretKey,
},
},
rmq_client.WithAwaitDuration(awaitDuration),
rmq_client.WithSubscriptionExpressions(map[string]*rmq_client.FilterExpression{
Topic: rmq_client.SUB_ALL,
}),
)
if err != nil {
log.Fatal(err)
}
// start simpleConsumer
err = simpleConsumer.Start()
if err != nil {
log.Fatal(err)
}
// graceful stop simpleConsumer
defer simpleConsumer.GracefulStop()
go func() {
fmt.Println("start receive message")
for {
mvs, err := simpleConsumer.Receive(context.TODO(), maxMessageNum, invisibleDuration)
if err != nil {
fmt.Println(err)
}
// ack message
for _, mv := range mvs {
fmt.Printf("message receive group:%s, timestamp:%d, offset:%d,body:%s\n",
*mv.GetMessageGroup(), time.Now().UnixMilli(), mv.GetOffset(), mv.GetBody())
if strings.EqualFold(*mv.GetMessageGroup(), "fifo001") {
time.Sleep(time.Second * 1)
}
fmt.Printf("message receive group:%s, timestamp:%d, offset:%d,body:%s,key=%v TO_ACK\n",
*mv.GetMessageGroup(), time.Now().UnixMilli(), mv.GetOffset(), mv.GetBody(), mv.GetKeys())
err = simpleConsumer.Ack(context.TODO(), mv)
if err != nil {
fmt.Println("simpleConsumer.Ack, err:", err)
}
}
fmt.Println()
}
}()
```
测试打印log如下：
```
consumer 1 log如下：
selectMessageQueueId: 1
message receive group:fifo001, timestamp:1743471018946, offset:5,body:message_5
message receive group:fifo001, timestamp:1743471019947, offset:5,body:message_5,key=[5] TO_ACK
selectMessageQueueId: 0
message receive group:fifo001, timestamp:1743471019984, offset:7,body:message_7
message receive group:fifo001, timestamp:1743471020985, offset:7,body:message_7,key=[7] TO_ACK
consumer2 log如下：
selectMessageQueueId: 0
message receive group:fifo001, timestamp:1743471019768, offset:6,body:message_6,key=[6]
message receive group:fifo001, timestamp:1743471020793, offset:6,body:message_6,key=[6] TO_ACK
selectMessageQueueId: 1
message receive group:fifo001, timestamp:1743471020806, offset:8,body:message_8,key=[8]
message receive group:fifo001, timestamp:1743471021819, offset:8,body:message_8,key=[8] TO_ACK
```
问题1 ：顺序消息顺序性
从文档看，[顺序消息负载均衡文档说明](https://rocketmq.apache.org/zh/docs/featureBehavior/08consumerloadbalance/)，“在消费过程中，前面的消息M1、M2被消费者Consumer A1处理时，只要消费状态没有提交，消费者A2是无法并行消费后续的M3、M4消息的，必须等前面的消息提交消费状态后才能消费后面的消息”。但是从日志看，M3，M4 在M2 未ACK前已经收到。这是由broker端处理由问题，还是我的用法不对？
```
consumer 2收到消息：timestamp:1743471019768, offset:6
consumer 1 收到消息：timestamp:1743471018946, offset:5，
consumer 1 准备ack 时刻： timestamp:1743471019947, offset:5,body:message_5,key=[5] TO_ACK
```
从这里看到consumer1 的offset5消息尚未ack，consumer2 已经收到了offset 6 的消息
问题2 ：queue 0 没有消息，client 轮询选择了queue0的时，也有消息返回，broker是怎么处理的？
GitHub link: https://github.com/apache/rocketmq/discussions/9306
----
This is an automatically sent email for dev@rocketmq.apache.org.
To unsubscribe, please send an email to: dev-unsubscribe@rocketmq.apache.org
Re: [D] FIFO topic 消费端非严格有序 [rocketmq]
Posted by "loringgit (via GitHub)" <gi...@apache.org>.
GitHub user loringgit added a comment to the discussion: FIFO topic 消费端非严格有序
1.topic创建的时候有指定是FIFO类型，创建命令：
```
./bin/mqadmin updateTopic -n nameserver(ip:port) -b mybroker(ip:port) -t test_fifo_topic -p 6 -r 2 -w 2 -a +message.type=FIFO
```
我用的是 github.com/apache/rocketmq-clients/golang/v5，NewSimpleConsumer默认使用的是v2.ClientType_SIMPLE_CONSUMER，有试过把这个默认值修改为 ClientType_PUSH_CONSUMER，测试结果没变化，你的意思是需要参考Java的PushConsumer自己实现OrderedMessage的Consumer吗？
2.日志中select MessageQueueId 是在ReceiveMessage方法增加的一行日志，代码见下面的注释
```
func (sc *defaultSimpleConsumer) Receive(ctx context.Context, maxMessageNum int32, invisibleDuration time.Duration) ([]*MessageView, error) {
if !sc.isOn() {
return nil, fmt.Errorf("simple consumer is not running")
}
if maxMessageNum <= 0 {
return nil, fmt.Errorf("maxMessageNum must be greater than 0")
}
sc.subscriptionExpressionsLock.RLock()
topics := make([]string, 0, len(sc.subscriptionExpressions))
for k := range sc.subscriptionExpressions {
topics = append(topics, k)
}
sc.subscriptionExpressionsLock.RUnlock()
// All topic is subscribed.
if len(topics) == 0 {
return nil, fmt.Errorf("there is no topic to receive message")
}
next := atomic.AddInt32(&sc.topicIndex, 1)
idx := utils.Mod(next+1, len(topics))
topic := topics[idx]
sc.subscriptionExpressionsLock.RLock()
filterExpression, ok := sc.subscriptionExpressions[topic]
sc.subscriptionExpressionsLock.RUnlock()
if !ok {
return nil, fmt.Errorf("no found filterExpression about topic: %s", topic)
}
subLoadBalancer, err := sc.getSubscriptionTopicRouteResult(ctx, topic)
if err != nil {
return nil, err
}
selectMessageQueue, err := subLoadBalancer.TakeMessageQueue()
if err != nil {
return nil, err
}
fmt.Printf("selectMessageQueueId: %v\n", selectMessageQueue.GetId()) // 打印QueueId
request := sc.wrapReceiveMessageRequest(int(maxMessageNum), selectMessageQueue, filterExpression, invisibleDuration)
timeout := sc.scOpts.awaitDuration + sc.cli.opts.timeout
return sc.receiveMessage(ctx, request, selectMessageQueue, timeout)
}
```
@francisoliverlee
GitHub link: https://github.com/apache/rocketmq/discussions/9306#discussioncomment-12696340
----
This is an automatically sent email for dev@rocketmq.apache.org.
To unsubscribe, please send an email to: dev-unsubscribe@rocketmq.apache.org