首先要配置Kafka服务监听外网ip的listeners
listeners=PLAINTEXT://内网IP:9092 advertised.listeners=PLAINTEXT://外网IP或域名:9092
我的Kafka集群内服务器配置了短域名host.com，所以外网设置需要多写后缀.host.com
[root@localhost kafka]# vi config/server.properties listeners=PLAINTEXT://hdss7-81:9092 advertised.listeners=PLAINTEXT://hdss7-81.host.com:9092
Python生产者代码
import json
import redis
import datetime
from kafka import KafkaProducer
from kafka.errors import KafkaError
class kafka_Producer():
def __init__(self, bootstrap_server, topic):
self.kafkabootstrap_server = bootstrap_server
self.kafkatopic = topic
self.kafkaproducer = KafkaProducer(bootstrap_servers=self.kafkabootstrap_server)
def sendjsondata(self, params):
try:
params_contents = json.dumps(params)
producer = self.kafkaproducer
future = producer.send(self.kafkatopic, params_contents.encode("utf-8"), None, None, 0) #通过回调函数获取发送信息
producer.flush() #调用此方法使所有缓冲记录立即可用来发送
recordmetadata = future.get(timeout=10)
print(recordmetadata, datetime.datetime.now().strftime('%Y%m%d%H%M%S'))
except KafkaError as e:
print(e)
# Press the green button in the gutter to run the script.
if __name__ == '__main__':
print('PyCharm')
bootstrap_servers = ["hdss7-81.host.com:9092", "hdss7-82.host.com:9092", "hdss7-83.host.com:9092"]
topicstr = "bigdata"
mykafkaproducer = kafka_Producer(bootstrap_servers, topicstr)
print("*" * 10, "生产者", "*" * 10)
for i in range(6, 10):
params = "{test}:{xxx}---" + str(i)
mykafkaproducer.sendjsondata(params)
print("end")
