我已经按照这里提到的教程重新设置了我的Kafka制作人:https://www.digitalocean.com/community/tutorials/how-to-install-apache-kafka-on-ubuntu-14-04
我正在使用cron和下面的脚本将一些事件推送到producer,服务器是ip:1.2.3.4
# !/usr/bin/env python
import threading, logging, time
import multiprocessing
import requests
import datetime
import json
from kafka import KafkaProducer
class CheckApis():
apisList = {"a": "https://test.eng.com/"}
kafkaProducer = "1.2.3.4:9092"
kafkaTopic = "sometopic"
producer = KafkaProducer(bootstrap_servers=kafkaProducer)
for key, value in apisList.items():
headers = {};
response = requests.request("GET", value, headers=headers)
message = {"app_name": key, "status": response.status_code, "message": "none", "timestamp": str(datetime.datetime.utcnow())}
producer.send(kafkaTopic, json.dumps(message).encode('utf-8'));
print (response.text)
print (response.status_code)
producer.close()
这很好,我可以使用以下命令查看推送的事件:
~/kafka/bin/kafka-console-consumer.sh --zookeeper 1.2.3.4:2181 --topic sometopic --from-beginning
但是,当我尝试从其他服务器(我的笔记本电脑)远程使用这些事件时,它失败并出现错误:
错误:侦听kafka事件时发生错误错误:connect econnrefused 5.6.7.8:9092(此处的某些不同ip不是1.2.3.4)
以下是我的用户代码(在节点js中使用kafka节点):
var ConsumerGroup = require('kafka-node').ConsumerGroup;
var healthConsumerOption = {
host: '1.2.3.4:2181',
autoCommit: true,
groupId: os.hostname(),
sessionTimeout: 15000,
protocol: ['roundrobin'],
fromOffset: 'latest'
};
var healthConsumerGroup = new ConsumerGroup(healthConsumerOption, healthTopics);
listenHealthEventsKafka: function(connections){
try{
healthConsumerGroup.on('error', onError);
healthConsumerGroup.on('message', onMessage);
healthConsumerGroup.on('connect', function(){
logger.info("Health consumer group is ready. ")
});
function onMessage(message){
var jsonData = JSON.parse(message.value);
console.log(message);
};
function onError(error){
logger.error("Some error occured while listening to kafka events " +error);
}
process.once('SIGNINT', function(){
async.each([healthConsumerGroup], function(consumer, callback){
logger.info("Closing the kafka health consumer process ");
consumer.close(true, callback);
});
})
}catch(error){
logger.error("Could not connect to kafka events for build " +error);
}
}
我是否需要在kafka服务器(server.properties)上进行其他配置以允许远程访问。。或者我做错了什么?请帮忙。
1条答案
按热度按时间vvppvyoh1#
关于你的Kafka经纪人
server.properties
你需要设置advertised.listeners
连接到外部ip,以便客户端可以正确连接到它。否则他们会尝试连接到内部ip(因为advertised.listeners
将默认为listeners
除非明确设置)裁判:https://kafka.apache.org/documentation/#brokerconfigs
另请参见https://www.confluent.io/blog/kafka-client-cannot-connect-to-broker-on-aws-on-docker-etc/