Apache Kafka流：无序消息

Question

我有一个Apache Kafka2.6制作人，它写的主题-A(TA)。我还有一个Kafka streams应用程序，它使用TA并写入topic-B（TB）。在streams应用程序中，我有一个自定义的时间戳提取器，它从消息负载中提取时间戳。

对于我的一个失败处理测试用例，我在应用程序运行时关闭了Kafka集群。

当生产者应用程序试图向TA写入消息时，它无法写入，因为集群已关闭，因此（我假设）缓冲了消息。假设它以递增的时间顺序接收4个消息m1、m2、m3、m4。（即m1是第一个，m4是最后一个）。

我从这里得到了一个解决方案，将所有事件从tA流到另一个中间主题（比如tA')，它将使用时间戳提取器到另一个主题。但我不确定这是否会导致事件根据提取的时间戳重新排序。

我的生产者代码如下所示（我使用Spring Cloud创建生产者）：Producer.java

@Service
public class Producer {

    private String topicName = "input-topic";
        
    private ApplicationProperties appProps;
    
    @Autowired
    private KafkaTemplate<String, MyEvent> kafkaTemplate;
    
    public Producer() {
        super();        
    }
    
    @Autowired
    public void setAppProps(ApplicationProperties appProps) {
        this.appProps = appProps;
        this.topicName = appProps.getInput().getTopicName();
    }

    public void sendMessage(String key, MyEvent ce) {
        ListenableFuture<SendResult<String,MyEvent>> future = this.kafkaTemplate.send(this.topicName, key, ce); 
        
    }
}

Answer 1

这是为什么？这是因为生产者中的缓冲是多线程的，每一个生产者同时对主题进行生产吗？

默认情况下，生产者允许最多5个并行的正在运行的请求到一个代理，因此，如果一些请求失败并被重试，请求顺序可能会改变。

若要避免此重新排序问题，您可以设置max.in.flight.requests.per.connection=1（可能会导致性能下降）或设置enable.idempotence=true。

顺便问一句：你没有说你的主题是有一个分区还是多个分区，你的消息是不是有一个密钥？如果您的主题有多个分区，并且您的消息被发送到不同的分区，那么读取时没有排序保证，因为偏移量排序只在一个分区内得到保证。

我假设自定义时间戳提取器将有助于在使用消息时对消息进行排序。但他们没有。或者也许我对时间戳提取器的理解是错误的。

时间戳提取器只提取一个时间戳。Kafka Streams不对任何消息重新排序，但总是以偏移量顺序处理消息。

我从这里得到了一个解决方案，将所有事件从tA流到另一个中间主题（比如tA')，它将使用时间戳提取器到另一个主题。但我不确定这是否会导致事件根据提取的时间戳重新排序。

不，它不会做任何重新排序。另一个SO问题是要更改时间戳，但如果您按照a、b、c顺序读取消息，那么结果将按照a、b、c顺序写入（只是使用不同的时间戳，但应保留偏移顺序）。

本讲座解释了一些更多细节:https://www.confluent.io/kafka-summit-San-francisco-2019/Whats-the-time-and-why/