FIFO topic 消费端非严格有序

[loringgit]

背景：
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 ：顺序消息顺序性
从文档看，顺序消息负载均衡文档说明，“在消费过程中，前面的消息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是怎么处理的？

[francisoliverlee]

问题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关键， 代码没有。 可以贴下实际的代码

[loringgit]

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

[francisoliverlee]

现在golang应该不支持fifo，grpc协议中的fifo没用被使用。

[guyinyou]

目前有一版 go pushconsumer的实现，里面有fifo的支持，感兴趣的可以一起review一下：
apache/rocketmq-clients#990