java - 学习卡夫卡0.8.2

Question

费了好大劲才找到一些新版本的简单入门示例。 KafkaProducer 之类的东西与 Producer 示例不同，网络上的许多旧代码似乎编译不一样。

有什么指导吗？ Apache Kafka 站点有零个 Java 生产者示例。

请指教。

Answer 1

在下面的示例中，我创建了一个使用 String 作为键并使用 byte[] 作为消息内容的生产者。

使用基本参数创建新的生产者:

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "server1:123,server2:456");
props.put(ProducerConfig.RETRIES_CONFIG, "3");
props.put(ProducerConfig.ACKS_CONFIG, "all");
props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "none");
props.put(ProducerConfig.BATCH_SIZE_CONFIG, 200);
props.put(ProducerConfig.BLOCK_ON_BUFFER_FULL_CONFIG, true);
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer");

KafkaProducer<String, byte[]> producer = new KafkaProducer<String, byte[]>(props);

同步发送消息:

producer.send(new ProducerRecord<>(topic, msgKey, msgContent)).get();

异步发送消息:

producer.send(new ProducerRecord<>(topic, msgKey, msgContent));

您的 Maven 依赖项对消费者和生产者都有好处。如果你只需要生产者，你可以使用:

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>0.8.2.0</version>
</dependency>

请注意，新的 Consumer API 可用但暂时无法使用。在代码源中，新API将返回null或抛出异常。