消息队列与中间件

你当前正在浏览历史版本 Vert.x 5.0.12 的文档。访问本页面的最新版本。

Vert.x Kafka 客户端

本组件提供了用于从 Apache Kafka 集群读取和向其发送消息的 Kafka 客户端。

作为消费者（Consumer），API 提供了订阅 Topic 分区以异步接收消息或将其作为流读取的方法（甚至支持暂停/恢复数据流）。

作为生产者（Producer），API 提供了向 Topic 分区发送消息的方法，体验如同向流中写入数据一样便捷。

在项目中引入 Vert.x Kafka 客户端

要在项目中使用本组件，请在构建描述文件的 dependencies 配置节中添加以下依赖：

Maven（在 pom.xml 中）：

<dependency>
  <groupId>io.vertx</groupId>
  <artifactId>vertx-kafka-client</artifactId>
  <version>5.0.12</version>
</dependency>

Gradle（在 build.gradle 文件中）：

compile "io.vertx:vertx-kafka-client:5.0.12"

创建 Kafka 客户端

创建消费者与生产者的过程非常相似，并且与原生 Kafka 客户端库的使用方式高度一致。

它们需要根据 Apache Kafka 官方文档中针对消费者（Consumer）以及生产者（Producer）的要求配置一组属性。

为此，可以配置一个包含这些属性的 Map，并将其传递给 KafkaConsumer 与 KafkaProducer 暴露的静态工厂方法：

Map<String, String> config = new HashMap<>();
config.put("bootstrap.servers", "localhost:9092");
config.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
config.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
config.put("group.id", "my_group");
config.put("auto.offset.reset", "earliest");
config.put("enable.auto.commit", "false");

// 使用 consumer 与 Apache Kafka 交互
KafkaConsumer<String, String> consumer = KafkaConsumer.create(vertx, config);

在上述示例中，通过 Map 实例创建了一个 KafkaConsumer 实例，指定了待连接的 Kafka 节点列表（示例中仅一个节点）以及用于从接收到的消息中解析 key 与 value 的反序列化器。

同样地，可以创建生产者：

Map<String, String> config = new HashMap<>();
config.put("bootstrap.servers", "localhost:9092");
config.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
config.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
config.put("acks", "1");

// 使用 producer 与 Apache Kafka 交互
KafkaProducer<String, String> producer = KafkaProducer.create(vertx, config);

创建 KafkaConsumer 的 EventLoop 线程将负责处理该消费者接收到的所有消息。例如，如果你希望在某个 Verticle 的 EventLoop 上处理消息，请务必在该 Verticle 的 start 方法内部创建 Kafka Consumer。

加入消费者组从 Topic 接收消息

为了开始从 Kafka Topic 接收消息，消费者可以使用 subscribe 方法订阅一组 Topic，作为消费者组的一部分（在创建时通过配置属性指定）。

也可以使用 subscribe 方法通过 Java 正则表达式动态匹配订阅多个 Topic。

你还需要使用 handler 注册处理入站消息的回调处理器：

consumer.handler(record -> {
  System.out.println("Processing key=" + record.key() + ",value=" + record.value() +
    ",partition=" + record.partition() + ",offset=" + record.offset());
});

// 使用列表订阅多个 topic
Set<String> topics = new HashSet<>();
topics.add("topic1");
topics.add("topic2");
topics.add("topic3");
consumer.subscribe(topics);

// 或使用 Java 正则表达式匹配订阅
Pattern pattern = Pattern.compile("topic\\d");
consumer.subscribe(pattern);

// 或仅订阅单个 topic
consumer.subscribe("a-single-topic");

处理器可以在调用 subscribe() 之前或之后注册；在两个方法都被调用之前，消息不会被真正消费。这允许你在订阅后先调用 seek()，最后再调用 handler()，以便仅从特定 Offset 开始消费消息。

在订阅时也可以传入结果处理器，以便在订阅操作完成时获得通知：

consumer.handler(record -> {
  System.out.println("Processing key=" + record.key() + ",value=" + record.value() +
    ",partition=" + record.partition() + ",offset=" + record.offset());
});

// 订阅多个 topic
Set<String> topics = new HashSet<>();
topics.add("topic1");
topics.add("topic2");
topics.add("topic3");
consumer
  .subscribe(topics)
  .onSuccess(v ->
    System.out.println("subscribed")
  ).onFailure(cause ->
    System.out.println("Could not subscribe " + cause.getMessage())
  );

// 或仅订阅单个 topic
consumer
  .subscribe("a-single-topic")
  .onSuccess(v ->
    System.out.println("subscribed")
  ).onFailure(cause ->
    System.out.println("Could not subscribe " + cause.getMessage())
  );

采用消费者组模式时，Kafka 集群会自动为消费者分配分区，并将同一消费者组中的其它在线消费者纳入考量，使得各分区能够在消费者之间均衡分布。

当某个消费者离开群组（其所分配的分区可被重新分配给其它消费者）或有新消费者加入群组时，Kafka 集群会自动进行分区重平衡（Re-balancing）。

你可以通过 partitionsRevokedHandler 与 partitionsAssignedHandler 在 KafkaConsumer 上注册监听器，以便在 Kafka 集群撤销或分配分区时收到通知：

consumer.handler(record -> {
  System.out.println("Processing key=" + record.key() + ",value=" + record.value() +
    ",partition=" + record.partition() + ",offset=" + record.offset());
});

// 注册分区分配与撤销的回调处理器
consumer.partitionsAssignedHandler(topicPartitions -> {
  System.out.println("Partitions assigned");
  for (TopicPartition topicPartition : topicPartitions) {
    System.out.println(topicPartition.getTopic() + " " + topicPartition.getPartition());
  }
});

consumer.partitionsRevokedHandler(topicPartitions -> {
  System.out.println("Partitions revoked");
  for (TopicPartition topicPartition : topicPartitions) {
    System.out.println(topicPartition.getTopic() + " " + topicPartition.getPartition());
  }
});

// 订阅 topic
consumer
  .subscribe("test")
  .onSuccess(v ->
    System.out.println("subscribed")
  ).onFailure(cause ->
    System.out.println("Could not subscribe " + cause.getMessage())
  );

在加入消费者组接收消息后，消费者可以随时通过 unsubscribe 退出消费者组以停止接收消息：

consumer.unsubscribe();

也可以传入结果回调处理器来捕获注销结果：

consumer
  .unsubscribe()
  .onSuccess(v ->
    System.out.println("Consumer unsubscribed")
  );

指定特定分区接收 Topic 消息

除了加入消费者组外，消费者还可以显式请求指定 Topic 的特定分区。当消费者不属于消费者组时，应用程序将无法依赖集群的自动重平衡机制。

你可以使用 assign 方法请求分配特定分区：

consumer.handler(record -> {
  System.out.println("key=" + record.key() + ",value=" + record.value() +
    ",partition=" + record.partition() + ",offset=" + record.offset());
});

// 指定待分配的分区集合
Set<TopicPartition> topicPartitions = new HashSet<>();
topicPartitions.add(new TopicPartition()
  .setTopic("test")
  .setPartition(0));

// 请求分配指定分区
consumer
  .assign(topicPartitions)
  .onSuccess(v -> System.out.println("Partition assigned"))
  // 分配完成后，获取分配给该消费者的分区列表
  .compose(v -> consumer.assignment())
  .onSuccess(partitions -> {
    for (TopicPartition topicPartition : partitions) {
      System.out.println(topicPartition.getTopic() + " " + topicPartition.getPartition());
    }
  });

与 subscribe() 类似，处理器可以在调用 assign() 之前或之后注册；在两者均被调用前不会开始消费消息。这允许你在调用 assign() 后先执行 seek()，最后再注册 handler()，以便仅从特定 Offset 开始消费。

调用 assignment 可以获取当前已分配的分区列表。

通过显式轮询（Poll）接收消息

除了依赖内部自动轮询机制接收消息外，客户端还可以订阅 Topic，不注册接收消息的 handler，而是通过 poll 方法显式拉取消息。

通过这种方式，应用程序可以完全自主掌控拉取消息的时机，例如在处理完上一批消息后再执行下一次拉取。

consumer
  .subscribe("test")
  .onSuccess(v -> {
    System.out.println("Consumer subscribed");

    // 每秒执行一次轮询拉取
    vertx.setPeriodic(1000, timerId ->
      consumer
        .poll(Duration.ofMillis(100))
        .onSuccess(records -> {
          for (int i = 0; i < records.size(); i++) {
            KafkaConsumerRecord<String, String> record = records.recordAt(i);
            System.out.println("key=" + record.key() + ",value=" + record.value() +
              ",partition=" + record.partition() + ",offset=" + record.offset());
          }
        })
        .onFailure(cause -> {
          System.out.println("Something went wrong when polling " + cause.toString());
          cause.printStackTrace();

          // 轮询异常时取消定时器停止轮询
          vertx.cancelTimer(timerId);
        })
    );
});

订阅成功后，应用程序启动一个周期性定时器，定期执行 poll 操作以从 Kafka 获取消息批次。

动态变更订阅或分区分配

在开始消费消息后，你可以随时通过重新调用 subscribe() 或 assign() 来变更所订阅的 Topic 或所分配的分区。

需要注意的是，由于内部存在消息缓冲机制，在 subscribe() 或 assign() 的完成回调被触发之后的一小段时间内，记录处理器（Record Handler）可能仍会观察到来自旧订阅或旧分区的消息。但批处理器（Batch Handler）则不会出现此现象：一旦完成处理器被调用，批处理器将仅观察到从新订阅或新分配分区读取的消息。

获取 Topic 分区信息

你可以调用 partitionsFor 获取指定 Topic 的分区信息：

consumer
  .partitionsFor("test")
  .onSuccess(partitions -> {
    for (PartitionInfo partitionInfo : partitions) {
      System.out.println(partitionInfo);
    }
  });

此外，listTopics 提供了集群中所有可用 Topic 及其关联的分区详情：

consumer
  .listTopics()
  .onSuccess(partitionsTopicMap ->
    partitionsTopicMap.forEach((topic, partitions) -> {
      System.out.println("topic = " + topic);
      System.out.println("partitions = " + partitions);
    })
  );

手动提交 Offset

在 Apache Kafka 中，消费者负责管理最后读取消息的 Offset（位移）。

默认情况下，每次从分区读取一批消息后会自动执行提交操作。创建消费者时配置参数 enable.auto.commit 必须设置为 true。

通过 commit 可以实现手动提交 Offset。这常用于实现至少一次（At-Least-Once）传递语义，确保读取的消息在真正处理完毕后才提交 Offset：

consumer.commit().onSuccess(v ->
  System.out.println("Last read message offset committed")
);

在 Topic 分区中重设位移（Seek）

Apache Kafka 能够长期持久化消息，消费者可以在 Topic 分区内部自由定位（Seek），实现对消息的任意位置访问。

你可以使用 seek 将读取 Offset 调整到指定位置：

TopicPartition topicPartition = new TopicPartition()
  .setTopic("test")
  .setPartition(0);

// seek 到指定 offset
consumer
  .seek(topicPartition, 10)
  .onSuccess(v -> System.out.println("Seeking done"));

当消费者需要从头重新读取流时，可以使用 seekToBeginning：

TopicPartition topicPartition = new TopicPartition()
  .setTopic("test")
  .setPartition(0);

// seek 到分区起始位置
consumer
  .seekToBeginning(Collections.singleton(topicPartition))
  .onSuccess(v -> System.out.println("Seeking done"));

最后，seekToEnd 可用于直接定位到分区的末尾：

TopicPartition topicPartition = new TopicPartition()
  .setTopic("test")
  .setPartition(0);

// seek 到分区末尾
consumer
  .seekToEnd(Collections.singleton(topicPartition))
  .onSuccess(v -> System.out.println("Seeking done"));

需要注意，由于存在内部消息缓冲，在 seek*() 方法的完成回调执行之后的一小段时间内，记录处理器可能仍会接收到从原 Offset 读取的缓冲消息。批处理器则不受此影响：一旦 seek*() 回调被调用，批处理器将仅观察到从新 Offset 读取的消息。

Offset 检索查询

你可以使用 Kafka 0.10.1.1 引入的 beginningOffsets API 获取给定分区的起始 Offset。与 seekToBeginning 不同，该方法不会改变消费者当前的 Offset。

Set<TopicPartition> topicPartitions = new HashSet<>();
TopicPartition topicPartition = new TopicPartition().setTopic("test").setPartition(0);
topicPartitions.add(topicPartition);

consumer
  .beginningOffsets(topicPartitions)
  .onSuccess(results ->
    results.forEach((topic, beginningOffset) ->
      System.out.println(
        "Beginning offset for topic=" + topic.getTopic() + ", partition=" +
          topic.getPartition() + ", beginningOffset=" + beginningOffset
      )
    )
  );

// 单分区查询便捷方法
consumer
  .beginningOffsets(topicPartition)
  .onSuccess(beginningOffset ->
    System.out.println(
      "Beginning offset for topic=" + topicPartition.getTopic() + ", partition=" +
        topicPartition.getPartition() + ", beginningOffset=" + beginningOffset
    )
  );

你可以使用 endOffsets API 获取给定分区的末尾 Offset。与 seekToEnd 不同，该操作不会改变消费者的当前 Offset。

Set<TopicPartition> topicPartitions = new HashSet<>();
TopicPartition topicPartition = new TopicPartition().setTopic("test").setPartition(0);
topicPartitions.add(topicPartition);

consumer.endOffsets(topicPartitions)
  .onSuccess(results ->
    results.forEach((topic, beginningOffset) ->
      System.out.println(
        "End offset for topic=" + topic.getTopic() + ", partition=" +
          topic.getPartition() + ", beginningOffset=" + beginningOffset
      )
    )
  );

// 单分区查询便捷方法
consumer
  .endOffsets(topicPartition)
  .onSuccess(endOffset ->
    System.out.println(
      "End offset for topic=" + topicPartition.getTopic() + ", partition=" +
        topicPartition.getPartition() + ", endOffset=" + endOffset
    )
);

你可以使用 offsetsForTimes API 根据时间戳查询 Offset，即搜索参数为纪元时间戳（Epoch Timestamp），调用返回写入时间戳大于等于指定时间戳的最小 Offset：

Map<TopicPartition, Long> topicPartitionsWithTimestamps = new HashMap<>();
TopicPartition topicPartition = new TopicPartition().setTopic("test").setPartition(0);

// 查询 60 秒前写入数据的 offset
long timestamp = (System.currentTimeMillis() - 60000);

topicPartitionsWithTimestamps.put(topicPartition, timestamp);
consumer
  .offsetsForTimes(topicPartitionsWithTimestamps)
  .onSuccess(results ->
    results.forEach((topic, offset) ->
      System.out.println(
        "Offset for topic=" + topic.getTopic() +
        ", partition=" + topic.getPartition() + "\n" +
        ", timestamp=" + timestamp + ", offset=" + offset.getOffset() +
        ", offsetTimestamp=" + offset.getTimestamp()
      )
    )
);

// 单分区查询便捷方法
consumer.offsetsForTimes(topicPartition, timestamp).onSuccess(offsetAndTimestamp ->
  System.out.println(
    "Offset for topic=" + topicPartition.getTopic() +
    ", partition=" + topicPartition.getPartition() + "\n" +
    ", timestamp=" + timestamp + ", offset=" + offsetAndTimestamp.getOffset() +
    ", offsetTimestamp=" + offsetAndTimestamp.getTimestamp()
  )
);

消息流控（暂停/恢复）

消费者可以控制入站消息流，对 Topic 的读取操作进行暂停和恢复。例如当需要更多时间处理当前消息时，可以先暂停消息流，随后恢复以继续处理。

你可以通过 pause 与 resume 实现此流控功能。

对于针对特定分区的暂停和恢复，在调用 pause() 方法的回调返回之后的一小段时间内，记录处理器可能仍会接收到来自该暂停分区的消息。批处理器则不受此影响：一旦 pause() 回调被调用，批处理器将仅观察到未暂停分区的消息。

TopicPartition topicPartition = new TopicPartition()
  .setTopic("test")
  .setPartition(0);

// 注册入站消息处理器
consumer.handler(record -> {
  System.out.println("key=" + record.key() + ",value=" + record.value() +
    ",partition=" + record.partition() + ",offset=" + record.offset());

  // 例如在读取分区 0 上直到 offset 5 的消息后暂停
  if ((record.partition() == 0) && (record.offset() == 5)) {

    // 暂停读取操作
    consumer.pause(topicPartition)
      .onSuccess(v -> System.out.println("Paused"))
      .onSuccess(v -> vertx.setTimer(5000, timeId ->
        // 5 秒后恢复读取
        consumer.resume(topicPartition)
      ));
  }
});

关闭消费者

调用 close 关闭消费者。关闭消费者将关闭所有打开的底层连接并释放所有消费者占用的资源。

关闭过程本身是异步的，在调用返回后可能仍需一段简短时间完成。如果你希望在关闭完全完成时获得通知，可以传入一个处理器：

consumer
  .close()
  .onSuccess(v -> System.out.println("Consumer is now closed"))
  .onFailure(cause -> System.out.println("Close failed: " + cause));

向 Topic 发送消息

你可以使用 write 向 Topic 发送消息（Record）。

最简单的发送方式是仅指定目标 Topic 与消息 Value，省略 Key 或分区；此时消息将以轮询（Round-Robin）方式均匀发送到该 Topic 的所有分区中：

for (int i = 0; i < 5; i++) {

  // 仅指定 topic 与消息内容，轮询写入目标分区
  KafkaProducerRecord<String, String> record =
    KafkaProducerRecord.create("test", "message_" + i);

  producer.write(record);
}

你也可以接收消息发送成功的元数据，例如目标 Topic、目标分区以及分配的 Offset：

for (int i = 0; i < 5; i++) {

  // 仅指定 topic 与消息内容，轮询写入目标分区
  KafkaProducerRecord<String, String> record =
    KafkaProducerRecord.create("test", "message_" + i);

  producer.send(record).onSuccess(recordMetadata ->
    System.out.println(
      "Message " + record.value() + " written on topic=" + recordMetadata.getTopic() +
      ", partition=" + recordMetadata.getPartition() +
      ", offset=" + recordMetadata.getOffset()
    )
  );
}

当需要将消息指定发送到特定分区时，可以传入分区编号或 Key：

for (int i = 0; i < 10; i++) {

  // 显式指定目标分区 0
  KafkaProducerRecord<String, String> record =
    KafkaProducerRecord.create("test", null, "message_" + i, 0);

  producer.write(record);
}

由于 Producer 根据 Key 的哈希值确定目标分区，因此你可以利用该特性确保具备相同 Key 的所有消息都被发送至同一分区，从而保证消息的严格顺序性：

for (int i = 0; i < 10; i++) {

  // 例如为奇偶消息指定不同的 key
  int key = i % 2;

  // 指定 key，相同 key 的消息将被发送到相同分区
  KafkaProducerRecord<String, String> record =
    KafkaProducerRecord.create("test", String.valueOf(key), "message_" + i);

  producer.write(record);
}

共享 Producer 在首次调用 createShared 时创建，其配置也在此时确定；后续复用共享 Producer 时必须使用完全一致的配置。

共享 Producer 实例

有时你希望在多个 Verticle 或不同上下文中共享同一个 Producer。

调用 KafkaProducer.createShared 返回一个可以安全共享的 Producer 实例：

KafkaProducer<String, String> producer1 = KafkaProducer.createShared(vertx, "the-producer", config);

// 后续可以安全关闭
producer1.close();

此方法返回的 Producer 之间会共享底层相同的系统资源（线程池、TCP 连接）。

当你使用完毕后只需调用关闭方法；当所有共享 Producer 均关闭后，底层资源会被完全自动释放。

关闭生产者

调用 close 关闭生产者。关闭生产者将关闭所有打开的连接并释放所有占用的资源。

关闭过程是异步的，在调用返回后可能仍需一段简短时间完成。如果你希望在完全关闭后获得通知，可以传入一个处理器：

producer
  .close()
  .onSuccess(v -> System.out.println("Producer is now closed"))
  .onFailure(cause -> System.out.println("Close failed: " + cause));

获取 Producer 端的 Topic 分区信息

你可以调用 partitionsFor 获取指定 Topic 的分区信息：

producer
  .partitionsFor("test")
  .onSuccess(partitions ->
    partitions.forEach(System.out::println)
  );

错误处理

Kafka 客户端（消费者或生产者）与 Kafka 集群之间的错误处理（例如超时）通过 exceptionHandler 或 exceptionHandler 进行注册：

consumer.exceptionHandler(e -> {
  System.out.println("Error = " + e.getMessage());
});

在 Verticle 中自动清理资源

如果你是在 Verticle 内部创建消费者与生产者的，当该 Verticle 被卸载（Undeploy）时，这些消费者和生产者将被自动关闭释放。

使用 Vert.x 序列化与反序列化器

Vert.x Kafka 客户端开箱即用地为 Buffer、JSON 对象与 JSON 数组提供了序列化器与反序列化器。

在消费者端可以使用 Buffer：

Map<String, String> config = new HashMap<>();
config.put("bootstrap.servers", "localhost:9092");
config.put("key.deserializer", "io.vertx.kafka.client.serialization.BufferDeserializer");
config.put("value.deserializer", "io.vertx.kafka.client.serialization.BufferDeserializer");
config.put("group.id", "my_group");
config.put("auto.offset.reset", "earliest");
config.put("enable.auto.commit", "false");

// 创建支持反序列化为 JSON 对象的消费者
config = new HashMap<>();
config.put("bootstrap.servers", "localhost:9092");
config.put("key.deserializer", "io.vertx.kafka.client.serialization.JsonObjectDeserializer");
config.put("value.deserializer", "io.vertx.kafka.client.serialization.JsonObjectDeserializer");
config.put("group.id", "my_group");
config.put("auto.offset.reset", "earliest");
config.put("enable.auto.commit", "false");

// 创建支持反序列化为 JSON 数组的消费者
config = new HashMap<>();
config.put("bootstrap.servers", "localhost:9092");
config.put("key.deserializer", "io.vertx.kafka.client.serialization.JsonArrayDeserializer");
config.put("value.deserializer", "io.vertx.kafka.client.serialization.JsonArrayDeserializer");
config.put("group.id", "my_group");
config.put("auto.offset.reset", "earliest");
config.put("enable.auto.commit", "false");

或者在生产者端：

Map<String, String> config = new HashMap<>();
config.put("bootstrap.servers", "localhost:9092");
config.put("key.serializer", "io.vertx.kafka.client.serialization.BufferSerializer");
config.put("value.serializer", "io.vertx.kafka.client.serialization.BufferSerializer");
config.put("acks", "1");

// 创建支持将 JSON 对象序列化的生产者
config = new HashMap<>();
config.put("bootstrap.servers", "localhost:9092");
config.put("key.serializer", "io.vertx.kafka.client.serialization.JsonObjectSerializer");
config.put("value.serializer", "io.vertx.kafka.client.serialization.JsonObjectSerializer");
config.put("acks", "1");

// 创建支持将 JSON 数组序列化的生产者
config = new HashMap<>();
config.put("bootstrap.servers", "localhost:9092");
config.put("key.serializer", "io.vertx.kafka.client.serialization.JsonArraySerializer");
config.put("value.serializer", "io.vertx.kafka.client.serialization.JsonArraySerializer");
config.put("acks", "1");

你也可以在创建时直接指定序列化器/反序列化器类型：

在消费者端：

Map<String, String> config = new HashMap<>();
config.put("bootstrap.servers", "localhost:9092");
config.put("group.id", "my_group");
config.put("auto.offset.reset", "earliest");
config.put("enable.auto.commit", "false");

// 创建支持反序列化 Buffer 的消费者
KafkaConsumer<Buffer, Buffer> bufferConsumer = KafkaConsumer.create(vertx, config, Buffer.class, Buffer.class);

// 创建支持反序列化 JSON 对象的消费者
KafkaConsumer<JsonObject, JsonObject> jsonObjectConsumer = KafkaConsumer.create(vertx, config, JsonObject.class, JsonObject.class);

// 创建支持反序列化 JSON 数组的消费者
KafkaConsumer<JsonArray, JsonArray> jsonArrayConsumer = KafkaConsumer.create(vertx, config, JsonArray.class, JsonArray.class);

或者在生产者端：

Map<String, String> config = new HashMap<>();
config.put("bootstrap.servers", "localhost:9092");
config.put("acks", "1");

// 创建支持序列化 Buffer 的生产者
KafkaProducer<Buffer, Buffer> bufferProducer = KafkaProducer.create(vertx, config, Buffer.class, Buffer.class);

// 创建支持序列化 JSON 对象的生产者
KafkaProducer<JsonObject, JsonObject> jsonObjectProducer = KafkaProducer.create(vertx, config, JsonObject.class, JsonObject.class);

// 创建支持序列化 JSON 数组的生产者
KafkaProducer<JsonArray, JsonArray> jsonArrayProducer = KafkaProducer.create(vertx, config, JsonArray.class, JsonArray.class);

RxJava 3 API

Kafka 客户端提供了原始 API 的响应式（Rxified）版本。

Observable<KafkaConsumerRecord<String, Long>> observable = consumer.toObservable();

observable
  .map(record -> record.value())
  .buffer(256)
  .map(
  list -> list.stream().mapToDouble(n -> n).average()
).subscribe(val -> {

  // 获取计算平均值

});

链路追踪上下文自动传播

当 Vert.x 启用了链路追踪（Tracing）时（参阅 setTracingOptions），追踪上下文将自动随 Kafka 消息进行传播。

Kafka 生产者在写入消息时会自动向链路添加 Span，追踪上下文将作为 Kafka 消息头（Headers）进行传播，消费者在读取消息时同样会自动创建对应的 Span。

遵循 OpenTracing 语义规范，相关的 Span Tag 包括：

span.kind：取值为 consumer 或 producer；
peer.address：可通过 setTracePeerAddress 进行配置；若未设置，则默认使用配置的 bootstrap servers；
peer.hostname：从 peer.address 解析得到；
peer.port：从 peer.address 解析得到；
peer.service：固定为 kafka；
message_bus.destination：设置为当前使用的 Topic。

Vert.x Kafka 管理客户端（Admin Client）

本组件为 Kafka Admin Client API 提供了 Vert.x 响应式封装。 Kafka Admin Client 用于创建、修改和删除 Topic，同时还提供了管理 ACL（访问控制列表）、消费者组（Consumer Groups）等资源的丰富方法。

创建 Kafka Admin 客户端

创建 Admin 客户端的方式与原生 Kafka 客户端库非常相似。

你需要按照 Apache Kafka 官方文档中 admin 配置说明所述配置一组属性。

为此，可以将包含相关配置属性的 Map 传递给 KafkaAdminClient 提供的静态工厂方法之一：

Properties config = new Properties();
config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

KafkaAdminClient adminClient = KafkaAdminClient.create(vertx, config);

列出所有 Topic

你可以调用 listTopics 列出集群中的所有 Topic。该方法接收用于处理结果的回调，回调参数中提供 Topic 列表：

adminClient.listTopics().onSuccess(topics ->
    System.out.println("Topics= " + topics)
);

查询 Topic 详情

你可以调用 describeTopics 查询集群中特定 Topic 的详细描述信息。查询 Topic 详情意味着获取诸如分区数量、副本数、Leader 节点、同步副本（ISR）等所有相关元数据。参数包括待查询的 Topic 名称列表，以及返回包含 Topic 名称与其对应 TopicDescription 的 Map 的结果处理器：

adminClient.describeTopics(Collections.singletonList("my-topic")).onSuccess(topics -> {
  TopicDescription topicDescription = topics.get("first-topic");

  System.out.println("Topic name=" + topicDescription.getName() +
      " isInternal= " + topicDescription.isInternal() +
      " partitions= " + topicDescription.getPartitions().size());

  for (TopicPartitionInfo topicPartitionInfo : topicDescription.getPartitions()) {
    System.out.println("Partition id= " + topicPartitionInfo.getPartition() +
      " leaderId= " + topicPartitionInfo.getLeader().getId() +
      " replicas= " + topicPartitionInfo.getReplicas() +
      " isr= " + topicPartitionInfo.getIsr());
  }
});

创建 Topic

你可以调用 createTopics 在集群中创建 Topic。参数为待创建的 Topic 列表以及处理结果的回调。待创建的 Topic 通过 NewTopic 类定义，指定名称、分区数及副本因子（Replication Factor）。你也可以显式描述副本分配（将每个副本映射到具体的 Broker ID），而不是直接指定分区数和副本因子（此时副本因子必须设为 -1）：

adminClient.createTopics(Collections.singletonList(new NewTopic("testCreateTopic", 1, (short)1)))
  .onSuccess(v -> {
    // Topic 创建成功
  })
  .onFailure(cause -> {
    // 创建 Topic 时发生异常
  });

删除 Topic

你可以调用 deleteTopics 在集群中删除 Topic。参数为待删除的 Topic 名称列表以及处理结果的回调：

adminClient.deleteTopics(Collections.singletonList("topicToDelete"))
  .onSuccess(v -> {
    // Topic 删除成功
  })
  .onFailure(cause -> {
    // 删除 Topic 时发生异常
  });

查询配置信息

你可以调用 describeConfigs 查询资源的配置信息。查询资源配置意味着获取集群资源（如 Topic 或 Broker）的所有配置详情。所需参数为你希望获取配置的资源列表，以及用于处理结果的回调。资源由 ConfigResource 集合描述，返回结果将每个资源映射到对应的 Config，其中包含每个配置参数的 ConfigEntry 项：

adminClient.describeConfigs(Collections.singletonList(
  new ConfigResource(org.apache.kafka.common.config.ConfigResource.Type.TOPIC, "my-topic"))).onSuccess(configs -> {
  // 检查配置项
});

修改配置信息

你可以调用 alterConfigs 修改资源的配置信息。修改资源配置意味着更新集群资源（如 Topic 或 Broker）的配置信息。所需参数为包含待更新配置的资源列表，以及用于处理结果的回调。你可以在单次调用中同时修改不同资源的配置。输入参数将每个 ConfigResource 映射为你希望应用的对应 Config：

ConfigResource resource = new ConfigResource(org.apache.kafka.common.config.ConfigResource.Type.TOPIC, "my-topic");
// 创建用于更新 Topic 上 retention.ms 取值的配置项
ConfigEntry retentionEntry = new ConfigEntry(TopicConfig.RETENTION_MS_CONFIG, "51000");
Map<ConfigResource, Config> updateConfig = new HashMap<>();
updateConfig.put(resource, new Config(Collections.singletonList(retentionEntry)));
adminClient.alterConfigs(updateConfig)
  .onSuccess(v -> {
    // 配置修改成功
  })
  .onFailure(cause -> {
    // 修改配置时发生异常
  });

列出所有消费者组

你可以调用 listConsumerGroups 列出集群中的所有消费者组。参数为处理结果的回调，回调中提供消费者组列表：

adminClient.listConsumerGroups().onSuccess(consumerGroups ->
  System.out.println("ConsumerGroups= " + consumerGroups)
);

查询消费者组详情

你可以调用 describeConsumerGroups 查询集群中特定消费者组的详细描述信息。查询消费者组详情意味着获取所有相关信息，如成员列表、成员 ID、订阅的 Topic、分区分配详情等。所需参数为待查询的消费者组名称列表，以及返回包含组名与对应 MemberDescription 的 Map 的结果处理器：

adminClient.describeConsumerGroups(Collections.singletonList("my-group")).onSuccess(groups -> {
  MemberDescription memberDescription = groups.get("first-member");
  // 处理成员描述信息
});

JDBC

AMQP