使用KafkaStreams处理主题时，如何从主题中获取上一条消息

Question

我一直在使用covid19api持有的数据实现Kafka生产者/消费者和流。

我试图从endpoint中提取每天的案例https://api.covid19api.com/all.然而，这个服务——以及这个API中的其他服务——拥有自疾病开始以来的所有数据（确诊、死亡和恢复病例），但积累了数据，而不是日常病例，这就是我最终要实现的。

使用transformValues和StoreBuilder（正如这里推荐的那样）对我也不起作用，因为场景不同。我使用transformValue功能实现了一些不同的功能，但每次检索到的前一个值都是主题的开头，而不是实际的前一个：

@Override
public String transform(Long key, String value) {
    String prevValue = state.get(key);
    log.info("{} => {}", key, value) ;
    if (prevValue != null) {
        Covid19StatDto prevDto = new Gson().fromJson(prevValue, Covid19StatDto.class);
        Covid19StatDto dto = new Gson().fromJson(value, Covid19StatDto.class);

        log.info("Current value {} previous {} ", dto.toString(), prevDto.toString());

        dto.setConfirmed(dto.getConfirmed() - prevDto.getConfirmed());

        String newDto = new Gson().toJson(dto);
        log.info("New value {}", newDto);
        return newDto;
    } else {
        state.put(key, value);
    }
    return value;
}

当我使用流处理主题时，如何从主题中获取之前的消息？任何帮助或建议都将不胜感激。

问候。

Answer 1

问题不仅仅是您只在状态存储中存储每个键的第一个值吗？如果在每个后续消息中，您始终希望看到前一条消息，那么您需要始终将当前消息存储在状态存储中，作为最后一步，对于exmaple：

@Override
public String transform(Long key, String value) {
    String prevValue = state.get(key);
    log.info("{} => {}", key, value) ;
    if (prevValue != null) {
        Covid19StatDto prevDto = new Gson().fromJson(prevValue, Covid19StatDto.class);
        Covid19StatDto dto = new Gson().fromJson(value, Covid19StatDto.class);

        log.info("Current value {} previous {} ", dto.toString(), prevDto.toString());

        dto.setConfirmed(dto.getConfirmed() - prevDto.getConfirmed());

        String newDto = new Gson().toJson(dto);
        log.info("New value {}", newDto);
        return newDto;
    }

    // Always update the state store:
    state.put(key, value);
    return value;
}