时间戳

首页 > 解决方案 > 使用 Spring & Lettuce 接收 Redis 流数据

redis - 使用 Spring & Lettuce 接收 Redis 流数据

问题描述

Spring boot每当 Redis 流附加新记录时，我都有以下代码来接收值。问题是接收者永远不会收到任何消息，而且订阅者在检查时subscriber.isActive()始终处于非活动状态。这段代码有什么问题？我错过了什么？文档供参考。

在spring boot启动时，初始化必要的redis资源

生菜连接工厂

@Bean
public RedisConnectionFactory redisConnectionFactory() {
    return new LettuceConnectionFactory("127.0.0.1", 6379);
}

来自连接工厂的 RedisTemplate

@Bean
public RedisTemplate<String, String> redisTemplate(RedisConnectionFactory connectionFactory) {
    RedisTemplate<String, String> redisTemplate = new RedisTemplate<>();
    redisTemplate.setConnectionFactory(connectionFactory);
    return redisTemplate;
}

休息控制器将数据附加到 redis 流

@PutMapping("/{name}")
public String post(@PathVariable String name) {
    return redisTemplate.opsForStream().add(StreamRecords.newRecord().in("streamx").ofObject(name)).getValue();
}

JMS 风格的命令式消息监听器

@Component
public class MyStreamListener implements StreamListener<String, MapRecord<String, String, String>> {

@Override
public void onMessage(MapRecord<String, String, String> message) {
    System.out.println("message received: " + message.getValue());
}

}

初始化监听器

  @Bean
public Subscription listener(MyStreamListener streamListener, RedisConnectionFactory redisConnectionFactory) throws InterruptedException {
    StreamMessageListenerContainer<String, MapRecord<String, String, String>> container = StreamMessageListenerContainer
            .create(redisConnectionFactory);
    Subscription subscription = container.receive(Consumer.from("my-group-1", "consumer-1"),
            StreamOffset.create("streamx", ReadOffset.latest())), streamListener);
    System.out.println(subscription.isActive()); // always false
    return subscription;
}

不过，我可以通过 api 附加到流中。

标签： redisspring-data-redis

解决方案

重要的一步是，StreamMessageListenerContainer订阅完成后开始。

container.start();

推荐阅读