1

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

消费者弹簧应用程序中出现的异常：-

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);
  }
}

第一次春季申请的制片人：-

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;
            }
        }
    }
}

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

4

1 回答 1

0

我遇到了这个错误，结果证明是与流记录对象_class在生产者和消费者之间不完全匹配（因为它使用完全限定的类名）有关的反序列化问题。@TypeAlias我可以通过在生产者端对象上添加 a 来修复它，以便它与消费者端相同对象的路径匹配：

@TypeAlias("com.example.consumer.model.MyStreamRecord")
public class MyStreamRecord {
...

由于 Redis 处理默认对象到哈希映射的方式（更多信息here和here） ，我还不得不在生产者和消费者端都使用aRedisTemplate<String,String>而不是。<String,MyStreamRecord>

于 2021-08-20T14:56:34.013 回答