为什么不能在Kafka上正确使用消费者？ kafka1 kafka2 kafka3 kafka1 kafka2 kafka3

如何解决为什么不能在Kafka上正确使用消费者？ kafka1 kafka2 kafka3 kafka1 kafka2 kafka3

有3个kafka服务器

在所有3台服务器上设置/etc/hosts。

192.168.0.1  kafka1
192.168.0.2  kafka2
192.168.0.3  kafka3

通过配置将Zookeeper和kafka安装在所有它们上

/usr/local/kafka_2.12-2.6.0/config/server.properties

kafka1

#
broker.id=1
listeners=PLAINTEXT://kafka1:9092
advertised.listeners=PLAINTEXT://kafka1:9092

kafka2

#
broker.id=2
listeners=PLAINTEXT://kafka2:9092
advertised.listeners=PLAINTEXT://kafka2:9092

kafka3

#
broker.id=3
listeners=PLAINTEXT://kafka3:9092
advertised.listeners=PLAINTEXT://kafka3:9092

启动zookeeper和kafka之后，创建一个新主题

[kafka@kafka1 ~]$ bin/kafka-topics.sh --create --zookeeper kafka1:2181,kafka2:2181,kafka3:2181 --replication-factor 1 --partitions 6 --topic topic1 --config cleanup.policy=delete --config delete.retention.ms=60000

检查所有三个节点上的群集状态

kafka1

[kafka@kafka1 ~]$ bin/kafka-topics.sh --describe --zookeeper kafka1:2181 --topic topic1
Topic: topic1   PartitionCount: 6   ReplicationFactor: 1    Configs: cleanup.policy=delete,delete.retention.ms=60000
    Topic: topic1   Partition: 0    Leader: 2   Replicas: 2 Isr: 2
    Topic: topic1   Partition: 1    Leader: 3   Replicas: 3 Isr: 3
    Topic: topic1   Partition: 2    Leader: 1   Replicas: 1 Isr: 1
    Topic: topic1   Partition: 3    Leader: 2   Replicas: 2 Isr: 2
    Topic: topic1   Partition: 4    Leader: 3   Replicas: 3 Isr: 3
    Topic: topic1   Partition: 5    Leader: 1   Replicas: 1 Isr: 1

kafka2

[kafka@kafka2 ~]$ bin/kafka-topics.sh --describe --zookeeper kafka1:2181 --topic topic1
Topic: topic1   PartitionCount: 6   ReplicationFactor: 1    Configs: cleanup.policy=delete,delete.retention.ms=60000
    Topic: topic1   Partition: 0    Leader: 2   Replicas: 2 Isr: 2
    Topic: topic1   Partition: 1    Leader: 3   Replicas: 3 Isr: 3
    Topic: topic1   Partition: 2    Leader: 1   Replicas: 1 Isr: 1
    Topic: topic1   Partition: 3    Leader: 2   Replicas: 2 Isr: 2
    Topic: topic1   Partition: 4    Leader: 3   Replicas: 3 Isr: 3
    Topic: topic1   Partition: 5    Leader: 1   Replicas: 1 Isr: 1

kafka3

[kafka@kafka3 ~]$ bin/kafka-topics.sh --describe --zookeeper kafka1:2181 --topic topic1
Topic: topic1   PartitionCount: 6   ReplicationFactor: 1    Configs: cleanup.policy=delete,delete.retention.ms=60000
    Topic: topic1   Partition: 0    Leader: 2   Replicas: 2 Isr: 2
    Topic: topic1   Partition: 1    Leader: 3   Replicas: 3 Isr: 3
    Topic: topic1   Partition: 2    Leader: 1   Replicas: 1 Isr: 1
    Topic: topic1   Partition: 3    Leader: 2   Replicas: 2 Isr: 2
    Topic: topic1   Partition: 4    Leader: 3   Replicas: 3 Isr: 3
    Topic: topic1   Partition: 5    Leader: 1   Replicas: 1 Isr: 1

但是当测试集群时，在kafka1上创建了一个生产器

[kafka@kafka1 ~]$ bin/kafka-console-producer.sh --broker-list kafka1:9092 --topic topic1
>

在其他任何节点（包括当前kafka1节点）上，运行consumer

bin/kafka-console-consumer.sh --bootstrap-server kafka1:9092 --topic topic1

遇到相同的错误：

# kafka1
[2020-08-21 02:09:57,299] WARN [Consumer clientId=consumer-console-consumer-39789-1,groupId=console-consumer-39789] Connection to node 2147483645 (kafka2/192.168.0.2:9092) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)
# kafka2
[2020-08-21 03:05:00,573] WARN [Consumer clientId=consumer-console-consumer-71891-1,groupId=console-consumer-71891] Connection to node -1 (kafka1/192.168.0.1:9092) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)
# kafka3
[2020-08-21 03:05:14,331] WARN [Consumer clientId=consumer-console-consumer-55574-1,groupId=console-consumer-55574] Connection to node -1 (kafka1/192.168.0.1:9092) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)

在/usr/local/kafka_2.12-2.6.0/config/server.properties中，我还尝试设置为

listeners=PLAINTEXT://localhost:9092
advertised.listeners=PLAINTEXT://localhost:9092

在所有3个节点上

。但是同样的问题。

解决方法

“侦听器”应代表生产者和使用者可以访问的节点FQDN和端口。 当生产者/消费者连接到Kafka时，就像在您的帖子中一样：

bin/kafka-console-consumer.sh --bootstrap-server kafka1:9092 --topic topic1

它打开到kafka1：9092的连接，然后从Kafka获取包含所有代理，它们的FQDN，PORT以及每个代理领导的主题分区的集群图。

您设法创建了使用Zookeeper完成且有效的主题。

根据您的WARN日志，“ broker1,2,3”被解析为192.168.0.1/2/3，因此IP解析正常。

尝试检查端口9092上的网络连接： 来自broker1运行：“ telnet broker2 9092”

或运行生产者命令以访问远程代理，即：

[kafka@**kafka1** ~]$ bin/kafka-console-producer.sh --broker-list **kafka2**:9092 --topic topic1

这将告诉您是否能够在端口9092上从broker1连接到broker2。