简介
kafka是最近几年很流行的消息队列中间件。在大数据以及后端服务领域有很广泛的应用。废话不多说，接下来直接上代码介绍python如何向kafka发送数据以及订阅数据。
消息发布
Kafka的消息是 " 发布--订阅" 模式的。 接下来先介绍向kakfa发布消息。先安装python的kafka连接模块。pip install kafka-python.
import time
from kafka import KafkaProducer
class KafkaMsgProducer:
def __init__(self, server):
self._server = server
self.producer = None
def connect(self):
if self.producer is None:
producer = KafkaProducer(bootstrap_servers=self._server)
self.producer = producer
def close(self):
if self.producer is not None:
self.producer.close()
self.producer = None
def send(self, topic, msg):
if self.producer is not None:
if not isinstance(msg, bytes):
msg = msg.encode("utf-8") # 将str类型转换为bytes类型
self.producer.send(topic=topic, value=msg)
def run():
producer = KafkaMsgProducer("localhost:9092")
producer.connect() # 建立连接
topic = "YanChampion-Test"
print("Start sending msg to kafka!")
for msg in "Hello! This is YanChampion speaking!".split():
producer.send(topic=topic, msg=msg) # 向kafka 指定topic发送数据
time.sleep(1)
if __name__ == '__main__':
run() # 运行发布消息程序
以上代码即可以向kafka指定topic发布消息了。注意，为了测试，先不运行producer.py
消息订阅
import time
from kafka import KafkaProducer
class KafkaMsgProducer:
def __init__(self, server):
self._server = server
self.producer = None
def connect(self):
if self.producer is None:
producer = KafkaProducer(bootstrap_servers=self._server)
self.producer = producer
def close(self):
if self.producer is not None:
self.producer.close()
self.producer = None
def send(self, topic, msg):
if self.producer is not None:
if not isinstance(msg, bytes):
msg = msg.encode("utf-8") # 将str类型转换为bytes类型
self.producer.send(topic=topic, value=msg)
def run():
producer = KafkaMsgProducer("localhost:9092")
producer.connect() # 建立连接
topic = "YanChampion-Test"
print("Start sending msg to kafka!")
for msg in "Hello! This is YanChampion speaking!".split():
producer.send(topic=topic, msg=msg) # 向kafka 指定topic发送数据
time.sleep(1)
if __name__ == '__main__':
run() # 运行程序
以上代码即可完成消息的订阅。
测试
因为kafka是 发布-定于模式。所以，一个topic可以有多个consumer订阅，并且，每个consumer都可以收到同一条消息。那么让我们先来运行两个consumer.py文件。
打开不同的终端
python3 consumer.py
接下来再运行producer.py
python3 producer.py
通过观察，最终可以看到，两个运行了consumer.py 的终端 都可以收到消息
[root@yanchampion kafka-demo]# python3 consumer.py
收到消息： b'Hello!'
收到消息： b'This'
收到消息： b'is'
收到消息： b'YanChampion'
收到消息： b'speaking!'
