使用Spring接收Redis streams数据

Question

我有下面的Spring boot代码，可以在Redis流附加新记录时接收值。问题是，当与subscriber进行检查时，接收方从未收到任何消息，订阅者也是如此。isActive（），始终处于非活动状态。这个代码有什么问题？我错过了什么？文件供参考。

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

生菜连接厂

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

从连接工厂重新创建模板

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

Rest控制器将数据附加到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将其附加到流中。

Answer 1

重要的一步是，在订阅完成后启动StreamMessageListenerContainer。

container.start();