如何检测无法访问的代理 Kafka 并重新连接？答案

【问题标题】：How to detect unreachable brokers Kafka and reconnect?如何检测无法访问的代理 Kafka 并重新连接？
【发布时间】：2018-10-17 05:06:24
【问题描述】：

方案： 2 代理 -2 主机 Java 客户端 - org.apache.kafka.clients.producer.KafkaProducer 或 org.apache.kafka.clients.consumer.KafkaConsumer Bootsrap 服务器属性包含 2 个 IP 地址（2 个代理） Zookeeper - 分布式协调器

构造函数 org.apache.kafka.clients.ClientUtil#parseAndValidateAddresses 的调用方法只检查没有验证实例的网络服务器（检查损坏的状态服务器或不启动）

【问题讨论】：

Kafka客户端不是一直在尝试自动重新连接吗？
不，它没有。获取第一个地址并抛出异常
可能是地址不可达。尽可能提供所有代理地址，以避免应用程序停机。
谢谢）是的，我使用了库 zookeeper for java。 stackoverflow.com/questions/29490113/... 用于列表代理，但实现了观察者存根（ new ZooKeeper("localhost:2181", 10000, new Watcher() - 没有空参数，因为在获取数据后抛出 NPE）；

标签： java apache-kafka apache-zookeeper

【解决方案1】：

默认情况下，Zookeeper 会跟踪所有连接的节点和可访问的代理。可以查询 ZK 看看目前连接了哪些节点。

# Connect to ZooKeeper
bin/zookeeper-shell.sh  localhost:2181
Connecting to localhost:2181
Welcome to ZooKeeper!
JLine support is disabled

WATCHER::

WatchedEvent state:SyncConnected type:None path:null
    >> ls /brokers/ids
        [1, 2, 3]
    >> ls /brokers/topics
        [myTopic, __consumer_offsets]
    >> get /brokers/ids/0
    Node does not exist: /brokers/ids/0
    >> get /brokers/ids/1
    {"listener_security_protocol_map":{"PLAINTEXT":"PLAINTEXT","SSL":"SSL"},"endpoints":["PLAINTEXT://172.31.4.17:9092","SSL://18.216.252.106:9093"],"jmx_port":9999,"host":"172.31.4.17","timestamp":"1522349056258","port":9092,"version":4}
    cZxid = 0x3000005f3
    ctime = Thu Mar 29 18:44:16 UTC 2018
    mZxid = 0x3000005f3
    mtime = Thu Mar 29 18:44:16 UTC 2018
    pZxid = 0x3000005f3
    cversion = 0
    dataVersion = 0
    aclVersion = 0
    ephemeralOwner = 0x2625d127c0b0003
    dataLength = 234
    numChildren = 0

【讨论】：

谢谢）是的，我使用了库 zookeeper for java。 stackoverflow.com/questions/29490113/… 用于列表代理，但实现了观察者存根（ new ZooKeeper("localhost:2181", 10000, new Watcher() - 没有空参数，因为在获取数据后抛出 NPE）；