如何使用Quarkus读取Redis Stream？官方暂不支持时的替代方案

Quarkus官方Redis扩展目前暂未提供Stream功能支持，以下是几种可行的替代实现方案：

方案1：优化Vert.x Redis客户端实现（适配持续消费） #

你的现有代码使用XRANGE仅能一次性查询历史数据，若要持续监听新消息，需改用XREAD的阻塞模式。优化后的代码示例：

@Startup
void onStart(@Observes StartupEvent ev) {
    Redis.createClient(vertx)
        .connect()
        .onSuccess(connection -> {
            System.out.println("已连接Redis: " + connection + " 线程: " + Thread.currentThread().getName());
            // 启动持续监听逻辑
            listenForStreamMessages(connection);
        })
        .onFailure(err -> System.err.println("Redis连接失败: " + err.getMessage()));
}

private void listenForStreamMessages(RedisConnection connection) {
    // 使用XREAD BLOCK模式永久阻塞，接收新消息（$表示仅监听当前时间点之后的消息）
    connection.send(Request.cmd(Command.XREAD)
            .arg("BLOCK").arg("0")
            .arg("STREAMS").arg("test").arg("$"))
        .onSuccess(res -> {
            System.out.println("收到Stream消息: " + res);
            // 递归调用保持持续监听
            listenForStreamMessages(connection);
        })
        .onFailure(err -> {
            System.err.println("读取Stream失败: " + err.getMessage());
            // 失败后延迟重试，避免中断消费
            vertx.setTimer(1000, id -> listenForStreamMessages(connection));
        });
}

关键说明：

• 若需消费历史数据，可将$替换为具体消息ID或0（表示从第一条消息开始）
• 添加失败重试逻辑，提升消费稳定性

方案2：引入第三方Redis客户端（如Lettuce） #

直接使用标准Redis客户端库绕过Quarkus扩展限制，以Lettuce为例：

1. 添加Maven依赖：

<dependency>
    <groupId>io.lettuce.core</groupId>
    <artifactId>lettuce-core</artifactId>
    <version>6.2.6.RELEASE</version>
</dependency>

2. 实现Stream消费逻辑：

@Startup
void onStart(@Observes StartupEvent ev) {
    RedisClient client = RedisClient.create("redis://localhost:6379");
    StatefulRedisConnection<String, String> connection = client.connect();
    RedisStreamCommands<String, String> streamCommands = connection.sync();
    
    // 持续阻塞消费新消息
    while (true) {
        List<StreamMessage<String, String>> messages = streamCommands.xread(
            XReadArgs.Builder.block(Duration.ofMillis(0)), 
            StreamOffset.from("test", "$")
        );
        for (StreamMessage<String, String> msg : messages) {
            System.out.println("消息ID: " + msg.getId() + " 内容: " + msg.getBody());
        }
    }
}

注意事项：

• 生产环境建议使用XREADGROUP实现消费组模式，避免重复消费
• 需通过@PreDestroy注解管理连接生命周期，防止资源泄漏

方案3：封装为Quarkus Reactive流（适配Reactive生态） #

将Redis Stream消费封装为Reactive流，整合Quarkus的响应式特性：

@Singleton
public class RedisStreamConsumer {
    private final RedisConnection connection;
    
    public RedisStreamConsumer(Vertx vertx) {
        // 同步等待连接建立，确保Bean初始化完成
        this.connection = Redis.createClient(vertx).connect().await();
    }
    
    public Multi<Response> consumeStream(String streamKey) {
        return Multi.createFrom().emitter(emitter -> {
            startListening(emitter, streamKey, "$");
        });
    }
    
    private void startListening(MultiEmitter<Response> emitter, String streamKey, String lastId) {
        connection.send(Request.cmd(Command.XREAD)
                .arg("BLOCK").arg("0")
                .arg("STREAMS").arg(streamKey).arg(lastId))
            .onSuccess(res -> {
                emitter.next(res);
                // 提取最后一条消息ID，继续监听后续消息
                String newLastId = res.get(0).get(1).get(0).get(0).toString();
                startListening(emitter, streamKey, newLastId);
            })
            .onFailure(emitter::fail);
    }
    
    @PreDestroy
    void closeConnection() {
        connection.close();
    }
}

业务代码中订阅Reactive流：

@Inject
RedisStreamConsumer consumer;

@Startup
void onStart(@Observes StartupEvent ev) {
    consumer.consumeStream("test")
        .subscribe().with(
            res -> System.out.println("收到Reactive消息: " + res),
            err -> System.err.println("消费失败: " + err.getMessage())
        );
}

内容的提问来源于stack exchange，提问作者Aswath Murugan