如何让RabbitMQ Stream消费者从最后一条消息开始消费（类MQTT保留消息）

要实现类似MQTT保留消息的效果（订阅后立即获取最新消息），你可以通过先查询流的精确末尾偏移量，再用该偏移量订阅的方式解决，具体步骤如下：

1. 获取流的最后一条消息偏移量 #

RabbitMQ Streams提供了API和CLI工具来查询流的当前状态，其中包含最后一条消息的偏移量：

代码方式（以Java客户端为例） #

使用StreamManager获取流信息，从中提取最后一条消息的偏移量：

import com.rabbitmq.stream.Environment;
import com.rabbitmq.stream.StreamInfo;
import com.rabbitmq.stream.StreamManager;

public class GetLastOffset {
    public static void main(String[] args) {
        try (Environment environment = Environment.builder().build()) {
            StreamManager streamManager = environment.streamManager();
            StreamInfo streamInfo = streamManager.getStreamInfo("your-stream-name");
            // lastOffset即为最后一条消息的偏移量
            long lastMessageOffset = streamInfo.getLastOffset();
            System.out.println("Last message offset: " + lastMessageOffset);
        }
    }
}

CLI工具方式 #

通过RabbitMQ官方的rabbitmq-streams命令行工具查询：

rabbitmq-streams stream info your-stream-name

在输出结果中找到last_offset字段，该值就是最后一条消息的精确偏移量。

2. 用精确偏移量订阅流 #

拿到最后一条消息的偏移量后，订阅时直接指定该偏移量，而不是使用last：

Java客户端示例 #

import com.rabbitmq.stream.Environment;
import com.rabbitmq.stream.MessageHandler;

public class SubscribeLatestMessage {
    public static void main(String[] args) {
        try (Environment environment = Environment.builder().build()) {
            StreamManager streamManager = environment.streamManager();
            long lastOffset = streamManager.getStreamInfo("your-stream-name").getLastOffset();
            
            environment.consumerBuilder()
                .stream("your-stream-name")
                // 指定最后一条消息的偏移量
                .offset(lastOffset)
                .messageHandler((context, message) -> {
                    // 处理最新消息
                    System.out.println("Received latest message: " + new String(message.getBodyAsBinary()));
                })
                .build();
            
            // 保持消费者运行
            Thread.currentThread().join();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

CLI工具示例 #

rabbitmq-streams consumer your-stream-name --offset <last_offset_value>

为什么last偏移量不符合需求？ #

RabbitMQ Streams的last偏移量指向的是最后一个消息块的起始位置，而非最后一条消息的位置。流中的消息是以块为单位批量存储的，所以用last订阅会读取整个最后块的所有消息（包括块内早于最后一条的消息），而通过上述方式获取的精确偏移量，能直接定位到最后一条消息，订阅后只会收到这条最新消息（以及后续新写入的消息）。

内容的提问来源于stack exchange，提问作者Philip Couling