Kafka 监听器详解

Kafka Assistant 是一款 Kafka GUI 管理工具——治理Broker,Topic,Group、查看生产详情、监控服务器状态、反对多种音讯格局。

[TOC]

你须要将 advertised.listeners（如果你应用Docker镜像，则为 KAFKA_ADVERTISED_LISTENERS）设置为内部地址（host/IP），以便客户端能够正确地连贯到它。否则，他们会尝试连贯到外部主机地址--如果无奈达到，问题就会接踵而来。

换一种说法，由Spencer Ruport提供。

listeners 是Kafka所绑定的接口。advertised.listeners 是客户端的连贯形式。

在这篇文章中，我将议论为什么这是有必要的配置 listeners 和 advertised.listeners，而后展现如何基于几个场景--Docker和AWS来做。

是谁在监听

Apache Kafka是一个分布式系统。数据是从一个给定的分区的领导者那里读取和写入的，这个领导者能够是集群中的任何一个Broker。当一个客户端（生产者/消费者）启动时，它将申请对于哪个broker是一个分区的领导者的元数据--它能够从任何broker那里做到这一点。返回的元数据将包含该分区的领导者broker的可用端点，而后客户端将应用这些端点连贯到broker，依据须要读/写数据。

正是这些端点给人们带来了麻烦。在单机上，运行裸机（没有虚拟机，没有Docker），可能只须要应用主机名（或只是localhost），这很容易。然而，一旦你进入更简单的网络设置和多节点，你就必须更加留神。

让咱们假如你有一个以上的网络。比方：

  • Docker外部网络（）加上主机
  • 云中的Broker（如AWS EC2）和本地的企业外部机器（甚至在另一个云中）。

你须要通知Kafka Broker如何互相分割，但也要确保内部客户端（生产者/消费者）能够分割到他们须要分割的Broker。

最要害的是，当你运行一个客户端时，你传递给它的Broker只是它要去的中央，并从那里取得集群中Broker的元数据。它为读/写数据而连贯的理论主机和IP是基于Broker在初始连贯中传递回来的数据--即便它只是一个繁多的节点，而且返回的Broker与它所连贯的Broker雷同。

为了正确配置，你须要理解Kafka Broker能够有多个监听器。一个监听器是一个组合：

  • Host/IP
  • Port
  • Protocol

让咱们来看看一些配置。通常状况下，协定也被用于监听器的名称，但在这里，让咱们通过应用监听器的形象名称来使它变得丑陋和清晰。

KAFKA_LISTENERS: LISTENER_BOB://kafka0:29092,LISTENER_FRED://localhost:9092KAFKA_ADVERTISED_LISTENERS: LISTENER_BOB://kafka0:29092,LISTENER_FRED://localhost:9092KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: LISTENER_BOB:PLAINTEXT,LISTENER_FRED:PLAINTEXTKAFKA_INTER_BROKER_LISTENER_NAME: LISTENER_BOB

我应用的是Docker配置名称--如果你间接配置 server.properties（例如，在AWS等），其对应名称在以下列表中缩进显示。

  • KAFKA_LISTENERS 是一个以逗号分隔的监听器列表，以及Kafka绑定监听的主机/IP和端口。对于更简单的网络，这可能是一个与机器上的特定网络接口相干的IP地址。默认是0.0.0.0，这意味着在所有接口上进行监听。

    • listeners

  • KAFKA_ADVERTISED_LISTENERS 是一个以逗号分隔的监听器列表，包含它们的主机/IP和端口。这是传回给客户端的元数据。

    • advertised.listeners

  • KAFKA_LISTENER_SECURITY_PROTOCOL_MAP 定义了每个监听器名称要应用的平安协定的键/值对。

    • listener.security.protocol.map

Kafka Broker之间的通信，通常是在外部网络（如Docker网络，AWS VPC等）。要定义应用哪个监听器，请指定 KAFKA_INTER_BROKER_LISTENER_NAME（inter.broker.listener.name）。应用的主机/IP必须是其他人能够从Broker机器上拜访的。

Kafka客户端很可能不在Broker的网络中，这就是额定监听器的作用。

每个监听器在被连贯到时，都会报告它能够被接连到的地址。你达到Broker的地址取决于所应用的网络。如果你从外部网络连接到经纪人，它将是一个不同于内部连贯的主机/IP。

当连贯到Broker时，返回给客户端的监听器将是你连贯到的监听器（基于端口）。

kafkacat是一个摸索这个问题的有用工具。应用 -L，你能够看到你所连贯的监听器的元数据。基于上述雷同的监听器配置（LISTENER_BOB/LISTENER_FRED），留神察看返回的后果：

  • 在9092端口（咱们映射为 LISTENER_FRED）进行连贯，Broker的地址被反馈为 localhost。

    $ kafkacat -b kafka0:9092 \       -LMetadata for all topics (from broker -1: kafka0:9092/bootstrap):1 brokers:broker 0 at localhost:9092

  • 在29092端口（咱们将其映射为 LISTENER_BOB）进行连贯，Broker的地址被反馈为kafka0。

    $ kafkacat -b kafka0:29092 \           -LMetadata for all topics (from broker 0: kafka0:29092/0):1 brokers:  broker 0 at kafka0:29092

你也能够用tcpdump来查看连贯到Broker服务器的客户端的流量，并发现从代理服务器返回的主机名。

为什么我能够连贯到Broker，但客户端依然失败？

即便你能与Broker建设初始连贯，在元数据中返回的地址依然可能是一个你的客户端无法访问的主机名。

让咱们一步一步来剖析。

  1. 咱们在AWS上有一个Broker。咱们想从咱们的笔记本电脑向它发送一个音讯。咱们晓得EC2实例的内部主机名（ec2-54-191-84-122.us-west-2.compute.amazonaws.com）。咱们曾经在平安组中创立了必要的条目，为咱们的入站流量关上Broker的端口。保险起见，还要查看咱们的本地机器是否能够连贯到AWS实例上的端口。

    $ nc -vz ec2-54-191-84-122.us-west-2.compute.amazonaws.com 9092found 0 associationsfound 1 connections:    1:  flags=82<CONNECTED,PREFERRED>  outif utun5  src 172.27.230.23 port 53352  dst 54.191.84.122 port 9092  rank info not available  TCP aux info available

    咱们连上了9092端口，持续

    echo "test"|kafka-console-producer --broker-list ec2-54-191-84-122.us-west-2.compute.amazonaws.com:9092 --topic test

    这一步产生了什么：

    • 咱们的笔记本电脑胜利地解析了 ec2-54-191-84-122.us-west-2.compute.amazonaws.com（到IP地址 54.191.84.122），并在9092端口连贯到AWS的机器。
    • Broker在端口9092上接支出站连贯。它把元数据返回给客户端，主机名是ip-172-31-18-160.us-west-2.compute.internal，因为这是Broker的主机名和监听器的默认值。

    • 而后，客户端试图应用它被赋予的元数据向代理发送数据。因为ip-172-31-18-160.us-west-2.compute.internal不能从互联网上解析，所以它失败了。

      $ echo "test"|kafka-console-producer --broker-list ec2-54-191-84-122.us-west-2.compute.amazonaws.com:9092 --topic test>>[2022-07-30 15:08:41,932] ERROR Error when sending message to topic test with key: null, value: 4 bytes with error: (org.apache.kafka.clients.producer.internals.ErrorLoggingCallback)org.apache.kafka.common.errors.TimeoutException: Expiring 1 record(s) for test-0: 1547 ms has passed since batch creation plus linger time

    • 然而，如果咱们从Broker所在的机器上尝试同样的事件。

      $ echo "foo"|kafka-console-producer --broker-list ec2-54-191-84-122.us-west-2.compute.amazonaws.com:9092 --topic test>>$ kafka-console-consumer --bootstrap-server ec2-54-191-84-122.us-west-2.compute.amazonaws.com:9092 --topic test --from-beginningfoo

      它工作得很好! 这是因为咱们正在连接到 9092 端口，它被配置为外部监听器，因而报告其主机名为 ip-172-31-18-160.us-west-2.compute.internal，这能够从Broker机器上解析（因为这是它本人的主机名！）。

    • 咱们能够通过应用kafkacat -L标记，看到Broker返回的元数据。

      $ kafkacat -b ec2-54-191-84-122.us-west-2.compute.amazonaws.com:9092 -LMetadata for all topics (from broker -1: ec2-54-191-84-122.us-west-2.compute.amazonaws.com:9092/bootstrap):1 brokers:  broker 0 at ip-172-31-18-160.us-west-2.compute.internal:9092

      很显著，外部主机名被返回。这也使得这个看似凌乱的谬误变得更有意义--连贯到一个主机名，在另一个主机上失去一个查问谬误。

      $ kafkacat -b ec2-54-191-84-122.us-west-2.compute.amazonaws.com:9092 -C -t test% ERROR: Local: Host resolution failure: ip-172-31-18-160.us-west-2.compute.internal:9092/0: Failed to resolve 'ip-172-31-18-160.us-west-2.compute.internal:9092': nodename nor servname provided, or not known

      这里，咱们在本地机器上应用消费者模式（-C）的kafkacat来尝试从主题中读取。和以前一样，因为咱们从元数据中的Broker那里失去了外部监听器的主机名，所以客户端无奈解析这个主机名来进行读/写。

批改我的hosts文件

我看到一个Stack Overflow的答案，倡议间接更新我的hosts文件......这不是更简略吗？
这能够解决问题，而不是真正修复它。

如果Broker服务器报告了一个客户端无奈连贯的主机名，那么在本地 /etc/hosts 中硬编码主机名/IP组合可能看起来是一个很好的修复。但这是一个十分软弱的、手动的解决方案。当IP发生变化时，当你挪动主机而遗记带配置hosts时，以及当其他人想做同样的事件时，会产生什么？

理解并理论修复你的网络的 advertised.listeners 设置要好得多。

如何连贯在Docker上的Kafka

为了在Docker中运行，你须要为Kafka配置两个监听器。

  • Docker网络内的通信。这可能是Broker之间的通信以及在Docker中运行的其余组件之间的通信，如Kafka Connect或第三方客户端或生产者。对于这些通信，咱们须要应用Docker容器的主机名。同一Docker网络上的每个Docker容器将应用Kafka Broker容器的主机名来达到它。

  • 非Docker网络流量。这可能是在Docker主机上本地运行的客户端，例如。假如他们将在localhost上连贯到从Docker容器裸露进去的端口。这是Docker Compose的片段。

    • Docker网络中的客户端应用监听器BOB进行连贯，端口为29092，主机名为kafka0。 这样做，他们会失去要连贯的主机名kafka0。每个Docker容器将应用Docker的外部网络来解析kafka0，并可能达到Broker。
    • Docker网络内部（但都在同一个Host Machine里）的客户端应用监听器FRED连贯，端口为9092，主机名为localhost。9092端口是由Docker容器裸露的，因而能够连贯到。当客户端连贯时，他们被赋予Broker元数据的主机名localhost，因而在读/写数据时连贯到此。
    • 上述配置无奈解决Docker内部和主机内部的客户端想要连贯的状况。这是因为无论是kafka0（Docker外部的主机名）还是localhost（Docker主机的环回地址）都是无奈解析的。

如何连贯到AWS/IaaS上的Kafka

我命名AWS是因为它是大多数人应用的，但这实用于任何IaaS/云解决方案。

这里实用的概念与Docker完全相同。次要的区别是，在Docker中，内部连贯很可能只是在localhost上（如上），而在云托管的Kafka中（如AWS上），内部连贯将来自非本地的机器。

另一个简单的问题是，尽管Docker网络与主机的网络重大隔离，但在IaaS上，内部主机名往往是能够在外部解析的，这使得你可能真正遇到这些问题时，一发不可收拾。

有两种办法，取决于你要连贯到Broker服务器的内部地址是否也能够在网络（如VPC）上的所有Broker服务器上进行本地解析。

选项1：内部地址是能够在本地解析的

在这里，你能够用一个监听器搞定。现有的监听器叫 PLAINTEXT，只须要设置 advertised.listeners（即传递给入站客户的那个）。

advertised.listeners=PLAINTEXT://ec2-54-191-84-122.us-west-2.compute.amazonaws.com:9092

当初外部和内部的连贯都将应用 ec2-54-191-84-122.us-west-2.compute.amazonaws.com 进行连贯。因为 ec2-54-191-84-122.us-west-2.com compute.amazonaws.com 既能够在本地也能够在内部解决，所以事件停顿顺利。

选项2：内部地址在本地无奈解析

你将须要为Kafka配置两个监听器。

  • AWS网络（VPC）内的通信。这可能是Broker之间的通信和在VPC中运行的其余组件之间的通信，如Kafka Connect或第三方客户端或生产商。对于这些通信，咱们须要应用EC2机器的外部IP（或主机名，如果配置了DNS）。
  • 内部AWS流量。这可能是测试来自笔记本电脑的连贯，或者只是来自不在亚马逊托管的机器。在这两种状况下，须要应用实例的内部IP（或主机名，如果配置了DNS）。

上面是一个例子：

listeners=INTERNAL://0.0.0.0:19092,EXTERNAL://0.0.0.0:9092listener.security.protocol.map=INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXTadvertised.listeners=INTERNAL://ip-172-31-18-160.us-west-2.compute.internal:19092,EXTERNAL://ec2-54-191-84-122.us-west-2.compute.amazonaws.com:9092inter.broker.listener.name=INTERNAL

原文地址：

  • Kafka Listeners – Explained