0

我使用KafkaEmbedded(and KafkaTemplate) 进行了单元测试，但消息顺序是随机的。有谁知道这是否合乎逻辑，是否可能是担保令？

这是我的代码：

public class KafkaTest {

  private static String TOPIC = "test.topic";

  @ClassRule
  public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, TOPIC);

  @Test
  public void testEmbeddedKafkaSendOrder() throws Exception {
    Map<String, Object> producerConfig = new HashMap<>();
    producerConfig.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, embeddedKafka.getBrokersAsString());
    producerConfig.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    producerConfig.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);

    KafkaTemplate<String, byte[]> kafkaTemplate = new KafkaTemplate<>(new DefaultKafkaProducerFactory<>(producerConfig));
    kafkaTemplate.send(TOPIC, "TEST1".getBytes()).get();
    kafkaTemplate.send(TOPIC, "TEST2".getBytes()).get();
    kafkaTemplate.send(TOPIC, "TEST3".getBytes()).get();
    kafkaTemplate.send(TOPIC, "TEST4".getBytes()).get();
    kafkaTemplate.send(TOPIC, "TEST5".getBytes()).get();

    Map<String, Object> consumerConfig = new HashMap<>();
    consumerConfig.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, embeddedKafka.getBrokersAsString());
    consumerConfig.put(ConsumerConfig.GROUP_ID_CONFIG, "consumer-test-group");
    consumerConfig.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    consumerConfig.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
    consumerConfig.put("auto.offset.reset", "earliest");

    final Consumer<String, byte[]> consumer = new KafkaConsumer<>(consumerConfig);
    embeddedKafka.consumeFromAnEmbeddedTopic(consumer, TOPIC);
    ConsumerRecords<String, byte[]> records = consumer.poll(100L);

    // Tests
    final Iterator<ConsumerRecord<String, byte[]>> recordIterator = records.iterator();
    while (recordIterator.hasNext()) {
      System.out.println("received:" + new String(recordIterator.next().value()));
    }
  }

例如，此代码打印（但顺序可以更改）：

received:TEST2
received:TEST4
received:TEST1
received:TEST3
received:TEST5
4

1 回答 1

2

在 Kafka 中，您可以确保消息的顺序在同一个分区上是相同的，但不是在主题上。

Note that as a topic typically has multiple partitions, there is
no guarantee of message time-ordering across the entire topic, just within a single
partition

从书中引述Kafka: The Definitive Guide: Real-Time Data and Stream Processing at Scale。您可以对此做些什么以及如何按顺序接收消息？选项1：

        kafkaTemplate.send(TOPIC,"1", "TEST1".getBytes()).get();
        kafkaTemplate.send(TOPIC,"1", "TEST2".getBytes()).get();
        kafkaTemplate.send(TOPIC,"1", "TEST3".getBytes()).get();
        kafkaTemplate.send(TOPIC,"1", "TEST4".getBytes()).get();
        kafkaTemplate.send(TOPIC,"1", "TEST5".getBytes()).get();

这样，对于每个值，您发送相同的键“1”。Kafka 将根据您的密钥选择分区。由于所有的键都是相等的，所有的消息都会进入同一个分区，你会按顺序收到你的记录。

选项 2：以这种方式初始化 KafkaEmbedded：

new KafkaEmbedded(1, true,1, TOPIC);

这样你就告诉 kafka 对于这个主题，你希望只有一个分区，所以每条记录都会进入那个分区。

于 2019-07-04T17:26:59.070 回答