Kafka Java Producer 无法向 kafka 实例发送消息答案

【问题标题】：Kafka Java Producer is not able to send message to kafka instanceKafka Java Producer 无法向 kafka 实例发送消息
【发布时间】：2020-09-02 16:57:36
【问题描述】：

我正在使用以下 docker-compose.yml 文件在 docker 容器中运行 kafka 实例。

version: "3"
services:
  zookeeper:
    image: 'bitnami/zookeeper:latest'
    ports:
      - '2181:2181'
    environment:
     - ALLOW_ANONYMOUS_LOGIN=yes
  kafka:
    image: 'bitnami/kafka:latest'
    ports:
      - '9092:9092'
      - '9093:9093'
    environment:
      - KAFKA_CFG_ZOOKEEPER_CONNECT=zookeeper:2181
      - ALLOW_PLAINTEXT_LISTENER=yes
      - KAFKA_BROKER_ID=1
      - KAFKA_CREATE_TOPICS="topic_name:1:3"
      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CLIENT:PLAINTEXT,EXTERNAL:PLAINTEXT
     - KAFKA_CFG_LISTENERS=CLIENT://:9092,EXTERNAL://:9093
     - KAFKA_CFG_ADVERTISED_LISTENERS=CLIENT://127.0.0.1:9092,EXTERNAL://127.0.0.1:9093
     - KAFKA_INTER_BROKER_LISTENER_NAME=EXTERNAL
    depends_on:
       - zookeeper

它运行得非常好。我通过kafkacat发送数据来测试kafka。没问题，我可以通过 kafka 消费者接收数据。请检查以下 kafkacat 命令。

kafkacat -P -b 127.0.0.1:9092 -t topic_name

kafkacat -C -b 127.0.0.1:9092 -t topic_name

但是，当我尝试通过 java 生产者代码发送它时，我无法从 kafkacat 消费者那里接收。请检查下面的 java 生产者代码。我想听听你的建议？提前致谢

public class DataProducer {

public static void main(String[] args) {
    KafkaTemplate<String,String> kafkaTemplate = new KafkaTemplate<>(new DefaultKafkaProducerFactory<>(kafkaConfig()));
    kafkaTemplate.send("topic_name", "test");
}

public static Map<String, Object> kafkaConfig() {
    Map<String, Object> props = new HashMap<>();

    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

    return props;
}}

在 kafkacat 命令输出之后添加元数据。

kafkacat -b 127.0.0.1:9092 -L
Metadata for all topics (from broker 1: 127.0.0.1:9092/1):
1 brokers:
broker 1 at 127.0.0.1:9092 (controller)
2 topics:
 topic "topic_name" with 1 partitions:
   partition 0, leader 1, replicas: 1, isrs: 1
topic "__consumer_offsets" with 50 partitions:
partition 0, leader 1, replicas: 1, isrs: 1
partition 1, leader 1, replicas: 1, isrs: 1
..
partition 48, leader 1, replicas: 1, isrs: 1
partition 49, leader 1, replicas: 1, isrs: 1

【问题讨论】：

kafkacat 是从您的机器运行还是从（其他）docker 容器运行？
我正在从我的电脑上运行 kafkacat 和 java 代码
你也可以添加这个命令的输出吗？ kafkacat -b 127.0.01:9092 -L
我加入了问题。

标签： java apache-kafka kafka-consumer-api kafka-producer-api

【解决方案1】：

代理配置似乎没问题，因为您取回了正确的元数据。

我认为问题出在您的代码中。 kafkaTemplate.send() 是一个异步操作，您的进程很可能在生产者设法实际发送消息之前结束。尝试将.get() 添加到该发送方法以强制它同步。

kafkaTemplate.send("topic_name", "test").get();

【讨论】：

【解决方案2】：

我有类似的问题。这很奇怪，但是 kafka-clients 版本的 maven 依赖项从 latest(2.6.0) 更改为 2.0.0（或 up to 2.5.0）帮助了我。

【讨论】：