我在一个spring-boot项目中有一个producer，它为同一个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);

第二个Spring应用中的耗电元件：-

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

请让我知道如果任何其他信息是必需的。任何帮助如何解决这个问题将不胜感激！