redis stream 消息自动确认

nunchan 2024-06-18 08:57

悬赏：5

rStreamTemplate.send(CommonRedisKey.ECANG_ORDER_UPDATE_QUEUE, CommonRedisKey.ECANG_ORDER_UPDATE_DATA_KEY, EcangOrderUpdateMsg.builder().orderId(RandomUtil.randomLong()).build());

@RStreamListener(name = CommonRedisKey.ECANG_ORDER_UPDATE_QUEUE, group = CommonRedisKey.ECANG_ORDER_UPDATE_GROUP, autoAcknowledge = false)
public void onMessage(Record> message){
    log.info("ecang order update listener, receive message:{}", JSON.toJSONString(message));
    final EcangOrderUpdateMsg value = message.getValue().get(CommonRedisKey.ECANG_ORDER_UPDATE_DATA_KEY);
    final Boolean updated = ecangService.updateEcangOrder(value.getOrderId());
    if (updated) {
        // 确认消息
        streamTemplate.acknowledge(CommonRedisKey.ECANG_ORDER_UPDATE_GROUP, message);
        log.info("ecang order update listener update order success, orderId:{}", value.getOrderId());
    }else{
        log.info("ecang order update listener update order fail, orderId:{}", value.getOrderId());
    }
}

• 

  nunchan (楼主)

  2024-06-20 10:01

  Redis的XREADGROUP命令有一个NOACK选项，添加NOACK选项则表示消息会自动被ACK。

  
  

  
  

   Redis5.0.7的XREAD命令是没有NOACK选项的。

  测试如下：

  127.0.0.1:6379> XREAD COUNT 1 NOACK STREAMS ecang:order:update $
  (error) ERR The NOACK option is only supported by XREADGROUP. You called XREAD instead.

  
  

  
  

  通过查看框架的源码： org.springblade.core.redis.stream.RStreamListenerDetector，类的这个方法：

  
  

  private void cluster(Consumer consumer, StreamOffset streamOffset, RStreamListener listener, Object bean, Method method) {
      boolean autoAcknowledge = listener.autoAcknowledge();
      StreamMessageListenerContainer.ConsumerStreamReadRequest readRequest = StreamMessageListenerContainer.StreamReadRequest.builder(streamOffset).consumer(consumer).autoAcknowledge(autoAcknowledge).build();
      StreamOperations opsForStream = redisTemplate.opsForStream();
      streamMessageListenerContainer.register(readRequest, (message) -> {
          // MapBackedRecord
          invokeMethod(bean, method, message, listener.readRawBytes());
          // ack
          // 当autoAcknowledge=false时，XREADGROUP不会添加NOACK选项，
          // 但是下面这个if判断又把消息给 ACK了
          if (!autoAcknowledge) {
              opsForStream.acknowledge(consumer.getGroup(), message);
          }
      });
  }

  
  

  
  

  所以，通过修改BladeX-Tool的源码，把上面cluster方法修改成如下即可解决问题：
  

  private void cluster(Consumer consumer, StreamOffset streamOffset, RStreamListener listener, Object bean, Method method) {
      boolean autoAcknowledge = listener.autoAcknowledge();
      StreamMessageListenerContainer.ConsumerStreamReadRequest readRequest = StreamMessageListenerContainer.StreamReadRequest.builder(streamOffset).consumer(consumer).autoAcknowledge(autoAcknowledge).build();
      StreamOperations opsForStream = redisTemplate.opsForStream();
      streamMessageListenerContainer.register(readRequest, (message) -> {
          // MapBackedRecord
          invokeMethod(bean, method, message, listener.readRawBytes());
          // ack
          // 修改if的条件
          if (autoAcknowledge) {
              opsForStream.acknowledge(consumer.getGroup(), message);
          }
      });
  }

  
  

  如果XREADGROUP自动ACK了消息，再调用XACK 也是没有问题的。

  经测试，可以多次XACK同一条消息，但只有第一条会返回成功

  127.0.0.1:6379> XACK ecang:order:update ecang:order:update:group 1718847213961-0
  (integer) 1
  127.0.0.1:6379> XACK ecang:order:update ecang:order:update:group 1718847213961-0
  (integer) 0
  127.0.0.1:6379> XACK ecang:order:update ecang:order:update:group 1718847213961-0
  (integer) 0

  
  

  回答: 2024-06-20 10:01

  学习Redis Stream 可以参考此篇博文：https://juejin.cn/post/7112825943231561741
  0 讨论(0)

  查看其它2个回答
  发布评论:

  提交评论

  ---

  •  加载中...

验证码

 看不清?

提交回复