从我的本地机器连接在 EC2 机器上运行的 Kafka

Question

我是 Kafka 新手，在论坛中搜索了不同的帖子，但找不到解决方案。我已经在 EC2 实例上安装了 kafka，并尝试从我的 ubuntu 本地机器上连接它。我的目标是让 python kafka 客户端（生产者和消费者）在我的本地机器上运行，并通过 EC2 kafka 实例发送/接收数据。这可能吗？

在 server.properties 配置文件中设置的属性：

listeners=PLAINTEXT://0.0.0.0:9092 
advertised.listeners=PLAINTEXT://<ec2-public-DNS>:9092

在 Kafka EC2 实例上：

netstat -an | grep LISTEN 
tcp 0 0 0.0.0.0:22 0.0.0.0:* LISTEN 
tcp6 0 0 :::9092 :::* LISTEN 

在 Kafka EC2 实例上的 Zookeeper cli 上：

get /brokers/ids/0
{"listener_security_protocol_map":{"PLAINTEXT":"PLAINTEXT"},"endpoints":["PLAINTEXT://<ec2-public-DNS>:9092"],"jmx_port":-1,"host":"<ec2-public-DNS>","timestamp":"1492900361516","port":9092,"version":4}
cZxid = 0xed
ctime = Sat Apr 22 22:32:41 UTC 2017
mZxid = 0xed
mtime = Sat Apr 22 22:32:41 UTC 2017
pZxid = 0xed
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x15b97cb9d060000
dataLength = 250
numChildren = 0

我本地机器上的 Python 客户端（生产者）：

from kafka import KafkaProducer
import time
import json

producer = KafkaProducer(bootstrap_servers="<ec2-public-DNS>:9092")
for i in range(100):
    dict = {}
    dict['name_'+str(i)] = 'FILE_' + str(i)
    dict['size_'+str(i)] = '23.' + str(i)
    dict['host_'+str(i)] = '10.0.0.0' + str(i)
    jd = json.dumps(dict)
    producer.send('console-test-topic', jd)
    time.sleep(2)

我本地机器上的 Python 客户端（消费者）：

from kafka import KafkaConsumer

consumer = KafkaConsumer('console-test-topic', bootstrap_servers="<ec2-public-DNS>:9092")
for msg in consumer:
    print (msg)

但是，生产者无法连接到 Kafka EC2 实例并且失败并出现以下错误：

**kafka.errors.NoBrokersAvailable: NoBrokersAvailable** 

我的安全组规则请参考链接：

goo.gl/ZUVknv

在我的本地机器上以调试模式运行 Producer：

DEBUG:kafka.producer.kafka:Starting the Kafka producer
DEBUG:kafka.metrics.metrics:Added sensor with name connections-closed
DEBUG:kafka.metrics.metrics:Added sensor with name connections-created
DEBUG:kafka.metrics.metrics:Added sensor with name select-time
DEBUG:kafka.metrics.metrics:Added sensor with name io-time
INFO:kafka.client:Bootstrapping cluster metadata from [('ec2-54-91-87-14.compute-1.amazonaws.com', 9092, 0)]
DEBUG:kafka.client:Attempting to bootstrap via node at ec2-54-91-87-14.compute-1.amazonaws.com:9092
DEBUG:kafka.metrics.metrics:Added sensor with name bytes-sent-received
DEBUG:kafka.metrics.metrics:Added sensor with name bytes-sent
DEBUG:kafka.metrics.metrics:Added sensor with name bytes-received
DEBUG:kafka.metrics.metrics:Added sensor with name request-latency
DEBUG:kafka.metrics.metrics:Added sensor with name node-bootstrap.bytes-sent
DEBUG:kafka.metrics.metrics:Added sensor with name node-bootstrap.bytes-received
DEBUG:kafka.metrics.metrics:Added sensor with name node-bootstrap.latency
DEBUG:kafka.client:Node bootstrap connected
DEBUG:kafka.cluster:Updated cluster metadata to ClusterMetadata(brokers: 1, topics: 2, groups: 0)
INFO:kafka.client:Bootstrap succeeded: found 1 brokers and 2 topics.
DEBUG:kafka.client:Initiating connection to node 0 at ec2-54-91-87-14.compute-1.amazonaws.com:9092
DEBUG:kafka.metrics.metrics:Added sensor with name node-0.bytes-sent
DEBUG:kafka.metrics.metrics:Added sensor with name node-0.bytes-received
DEBUG:kafka.metrics.metrics:Added sensor with name node-0.latency
INFO:kafka.producer.kafka:Kafka producer closed
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/usr/local/lib/python2.7/dist-packages/kafka/producer/kafka.py", line 335, in __init__
    **self.config)
  File "/usr/local/lib/python2.7/dist-packages/kafka/client_async.py", line 210, in __init__
    self.config['api_version'] = self.check_version(timeout=check_timeout)
  File "/usr/local/lib/python2.7/dist-packages/kafka/client_async.py", line 828, in check_version
    raise Errors.NoBrokersAvailable()
kafka.errors.NoBrokersAvailable: NoBrokersAvailable

我尝试在另一个 EC2 实例中运行生产者客户端（在与 kafka 实例相同的 VPN 中），它工作正常。但是，当生产者在我的本地机器上运行时，它不起作用。 'advertised.listeners' 属性是否在同一个（AWS VPN）网络中宣传 kafka 经纪人？或者我也可以从我的本地机器连接它？如果有人能指出正确的方向，请告诉我。

Answer 1

几个月前我经历了类似的情况，基本上我在弗吉尼亚北部有一个带有 Kafka 的 ec2 实例，我在本地机器上配置了 topbeat 以将指标发送到该 ec2 实例。 我能够通过添加

让它工作

advertised.host.name=public-ip

作为 kafka 的 server.properties 中的配置，但根据documentation，此属性已被弃用。

进一步阅读documentation，据说如果你在IaaS环境中，你必须配置与broker绑定的接口不同的advertised.listeners。

Answer 2

您是否疏忽了将 AWS 安全组中的入站端口 9092 打开到本地网络的外部 IP？如果您选择向所有人开放，请将其打开至0.0.0.0/0（但请注意安全隐患）。