【Python-Kafka】Python毗连Windows下的kafka服务器，创建生产者Producer和 ...

---

一、Windows11本地开启Kafka服务器

1.1、开启Zookeeper

Windows下安装Zookeeper的安装可以参考这篇博文：
Windows下安装Zookeeper（图文记录详细步调，手把手包安装成功）
先启动Zookeeper服务
管理员权限打开命令窗口，输入命令zkServer
，启动Zookeeper服务：

1. zkServer

复制代码

1.2、开启Kafka服务

Windows下安装Kafka的安装可以参考这篇博文：
【Kafka】Windows下安装Kafka（图文记录详细步调）
管理员权限打开命令窗口，进入到Kafka安装目录（D:\bigdata\kafka\2.12-3.5.1）。
输入如下命令启动Kafka服务：

1. .\bin\windows\kafka-server-start.bat .\config\server.properties

复制代码

二、Python创建生产者Producer

起首运行生产者Producer
KafkaConsumerDemo.py

1. #!/usr/bin/env python
2. # -*- coding: utf-8 -*-
3. import json
4. import time
6. from kafka import KafkaProducer
7. from kafka.errors import KafkaError
9. producer = KafkaProducer(bootstrap_servers=['127.0.0.1:9092'])
10. # Assign a topic
11. topic = 'test'
14. def test():
15.     print('begin')
16.     n = 1
17.     # 初始化一个变量用来存储用户输入
18.     user_input = ""
19.     try:
20.         # 使用while循环来持续获取用户输入
21.         while user_input != "quit":  # 条件是用户输入不是'quit'
22.             user_input = input("Kafka生产者，生产消息，输入内容（输入'quit'退出）：")
23.             if user_input != "quit" and user_input != "":
24.                 producer.send(topic, user_input.encode())
25.                 n += 1
26.                 # time.sleep(1.5)
27.     except KafkaError as e:
28.         print(e)
29.     finally:
30.         producer.close()
31.         print('done')
34. def test_json():
35.     msg_dict = {
36.         "sleep_time": 10,
37.         "db_config": {
38.             "database": "test_1",
39.             "host": "xxxx",
40.             "user": "root",
41.             "password": "root"
42.         },
43.         "table": "msg",
44.         "msg": "Hello World"
45.     }
46.     msg = json.dumps(msg_dict)
47.     producer.send(topic, msg, partition=0)
48.     producer.close()
51. if __name__ == '__main__':
52.     test()

复制代码

可以不绝发送消息：

三、Python创建消费者Consumer

再创建消费者Consumer
KafkaConsumerDemo.py

1. #!/usr/bin/env python
2. # -*- coding: utf-8 -*-
3. from pykafka import KafkaClient
4. from pykafka.common import OffsetType
6. bootstrap_servers = '127.0.0.1:9092'
7. group_id = 'test'
10. class KConsumer(object):
11.     """kafka 消费者; 动态传参,非配置文件传入;
12.       kafka 的消费者应该尽量和生产者保持在不同的节点上;否则容易将程序陷入死循环中;
13.      """
14.     _encode = "UTF-8"
16.     def __init__(self, topics, bootstrap_server=None, group_id=group_id, partitions=None):
17.         """ 初始化kafka的消费者;
18.            1. 设置默认 kafka 的主题, 节点地址, 消费者组 id(不传入的时候使用默认的值)
19.            2. 当需要设置特定参数的时候可以直接在 kwargs 直接传入,进行解包传入原始函数;
20.          Args:
21.            topics: str; kafka 的消费主题;
22.            bootstrap_server: list; kafka 的消费者地址;
23.            group_id: str; kafka 的消费者分组 id,默认是 start_task 主要是接收并启动任务的消费者,仅此一个消费者组id;
24.          """
25.         if bootstrap_server is None:
26.             bootstrap_server = bootstrap_servers
27.         self.client = KafkaClient(hosts=bootstrap_server)
28.         # 选择要消费的topic
29.         vpn_topic = self.client.topics[topics]
30.         self.consumer = vpn_topic.get_simple_consumer(consumer_group=group_id,
31.                                                       consumer_timeout_ms=200,
32.                                                       auto_commit_enable=False,  # 自动提交偏移量
33.                                                       # LATEST 获取当前偏移量最新消息  EARLIEST 从头开始获取信息
34.                                                       auto_offset_reset=OffsetType.EARLIEST)
36.     def recv(self):
37.         """
38.          接收消费中的数据
39.          Returns:
40.          """
41.         return self.consumer
44. def main():
45.     """
46.     kafka消费队列入口
47.     :param topic:
48.     :return:
49.     """
50.     obj = KConsumer(topics="test")
51.     while True:
52.         for message in obj.recv():
53.             # data = eval(message.value.decode('utf-8'))
54.             data = message.value.decode('utf-8')
55.             print("Kafka消费者接受到的消息：" + data)
58. if __name__ == '__main__':
59.     main()

复制代码

可以发现可以或许实时收到生产者发送来的消息：

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。