Redis 消费者无法消费特定事件

问题描述

我在 spring boot 项目中有一个生产者，它为同一个 redis 消费者组和一个消费相同的消费者（另一个 spring boot 项目）生产两个事件。 redis 消费者只能消费一种类型的事件。我在使用第二种类型的事件时看到以下异常：-

在消费者 Spring 应用程序中出现异常：-

2021-01-06 19:14:22,580 ERROR [SimpleAsyncTaskExecutor-1] DefaultStreamMessageListenerContainer$LoggingErrorHandler: Unexpected error occurred in scheduled task.
java.lang.IllegalArgumentException: Value must not be null!
    at org.springframework.util.Assert.notNull(Assert.java:198)
    at org.springframework.data.redis.connection.stream.Record.of(Record.java:81)
    at org.springframework.data.redis.connection.stream.MapRecord.toObjectRecord(MapRecord.java:147)
    at org.springframework.data.redis.core.StreamObjectMapper.toObjectRecord(StreamObjectMapper.java:132)
    at org.springframework.data.redis.core.StreamObjectMapper.map(StreamObjectMapper.java:158)
    at org.springframework.data.redis.core.StreamOperations.read(StreamOperations.java:377)
    at org.springframework.data.redis.stream.DefaultStreamMessageListenerContainer.lambda$getReadFunction$2(DefaultStreamMessageListenerContainer.java:232)
    at org.springframework.data.redis.stream.StreamPollTask.doLoop(StreamPollTask.java:138)
    at org.springframework.data.redis.stream.StreamPollTask.run(StreamPollTask.java:123)
    at java.lang.Thread.run(Thread.java:748)

Redis 流任务:-

@Component
public class RedisStreamTask<T> {

  @Autowired
  private RedisTemplate<String,Object> redisTemplate;

  @Async
  public void publish(final StreamingEvent<T> event) {

    final ObjectRecord<String,T> record = StreamRecords.newRecord()
        .ofObject(event.getMessage())
        .withStreamKey(event.getKey());

    final RecordId recordId = redisTemplate
        .opsForStream()
        .add(record);
  }
}

第一个 spring 应用程序中的生产者：-

final StreamingEvent<MyBean> streamingEvent = new StreamingEvent<>();
streamingEvent.setKey("CONSUMER_GROUP_KEY");
streamingEvent.setMessage(myBean);
redisStreamTask.publish(streamingEvent);

第二次春季应用中的消费者：-

imports...

@Component
public class Consumer implements StreamListener<String,ObjectRecord<String,MyBean>>  {

    private AtomicInteger atomicInteger = new AtomicInteger(0);

    private final Logger log = LoggerFactory.getLogger(Consumer.class);

    @Autowired
    private ReactiveRedisTemplate<String,String> redisTemplate;

    @Autowired
    public RedisConfig config;

    @Autowired
    private MyService myService;


    @Override
    public void onMessage(ObjectRecord<String,MyBean> record) {
        record.getValue();
        tempCreateFile.log(record.getValue());

        try {
            myService.saveEvent(record.getValue());
        } catch (Exception e) {
            log.error("Error while storing redis events",e);
        }
        atomicInteger.incrementAndGet();
    }

    @Scheduled(fixedRate = 10000)
    public void showPublishedEventsSoFar() {
        log.info("Total Consumed :: " + atomicInteger.get());
    }

    @PostConstruct
    public void afterPropertiesSet() throws Exception {
        try {
            log.info("Create Consumer group: {}",config.getConsumerGroupName());
            redisTemplate.opsForStream().createGroup(config.getStreamName(),config.getConsumerGroupName()).block();
        } catch (RedisSystemException e) {
            if (e.getRootCause().getClass().equals(RedisBusyException.class)) {
            } else if (e.getRootCause().getClass().equals(RedisCommandExecutionException.class)) {
                log.info("STREAM - Stream does not yet exist,creating empty stream: " + config.getStreamName());
                redisTemplate.opsForStream().add( config.getStreamName(),Collections.singletonMap("","")).block();
                redisTemplate.opsForStream().createGroup(config.getStreamName(),config.getConsumerGroupName()).block();
            } else {
                e.printStackTrace();
                throw e;
            }
        }
    }
}

如果需要任何其他信息，请告诉我。任何有关如何解决此问题的帮助将不胜感激！

解决方法

暂无找到可以解决该程序问题的有效方法，小编努力寻找整理中！

如果你已经找到好的解决方法，欢迎将解决方案带上本链接一起发送给小编。

小编邮箱:dio#foxmail.com (将#修改为@）