一对kafka消息生产者-消息消费者，python

水深无声 2023-09-27 08:40 57阅读 0赞

## 一对kafka消息生产者-消息消费者，python ##

消息生产者：

import datetime
    import random
    
    from kafka import KafkaProducer
    import json
    
    from loguru import logger
    import kafka_config
    
    
    def on_send_success(record_metadata=None):
        print(record_metadata.topic)
        print(record_metadata.partition)
        print(record_metadata.offset)
    
    
    def on_send_error(excp=None):
        logger.error('失败', exc_info=excp)
    
    
    if __name__ == '__main__':
        msg_producer = KafkaProducer(bootstrap_servers=f'{kafka_config.server_host}:{kafka_config.server_port}',
                                     value_serializer=lambda v: json.dumps(v).encode('utf-8'))
    
        val = {
            'event_type': 0,
            'msg': random.randint(0, 100),
            'time': datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')
        }
    
        future = msg_producer \
            .send(topic=r'some_topic', key=b'some_key', value=json.dumps(val)) \
            .add_callback(on_send_success) \
            .add_errback(on_send_error)
    
        msg_producer.flush()
    
        result = future.get(timeout=10)
        print(result)

消息消费者：

import json
    
    from kafka import KafkaConsumer
    
    import kafka_config
    
    if __name__ == '__main__':
        consumer = KafkaConsumer(bootstrap_servers=[f'{kafka_config.server_host}:{kafka_config.server_port}'],
                                 group_id='my_group')
        conn = consumer.bootstrap_connected()
        if conn:
            print('连接OK')
    
        topics = consumer.topics()
        for t in topics:
            print(t)
    
        consumer.subscribe(topics=['some_topic'])
    
        for msg in consumer:
            print(msg.topic, json.loads(msg.value))
            print("%s:%d:%d: key=%s value=%s" % (msg.topic, msg.partition, msg.offset, msg.key, msg.value))

kafka配置文件，kafka\_config.py：

server_host = '127.0.0.1'
    server_port = 9092

先启动消费者，然后生产者再发送消息。Kafka的broker配置参考:

[kafka连接zookeeper配置搭建环境、安装和启动，Windows环境下\_zhangphil的博客-CSDN博客\_kafka zookeeper配置kafka，分布式集群架构下，高性能的流式事件数据（主要是消息）集成、发布（生产）和订阅（分发、消费）组件（中间件）。kafka依赖zooeeper（数据后端），这里有Windows下安装配置启动zookeeper的 文章（1）：Windows 10环境zookeeper单机伪集群部署和配置\_Zhang Phil-CSDN博客Windows 10环境zookeeper单机伪集群部署和配置1，首先到zookeeper项目主页地址下载项目包，https://archive.apache.org/dist![favicon32.ico][]https://blog.csdn.net/zhangphil/article/details/123086790][kafka_zookeeper_Windows_zhangphil_-CSDN_kafka zookeeper_kafka_kafka_zooeeper_Windows_zookeeper_ _1_Windows 10_zookeeper_Zhang Phil-CSDN_Windows 10_zookeeper_1_zookeeper_https_archive.apache.org_dist_favicon32.ico_https_blog.csdn.net_zhangphil_article_details_123086790]

[favicon32.ico]: https://g.csdnimg.cn/static/logo/favicon32.ico
[kafka_zookeeper_Windows_zhangphil_-CSDN_kafka zookeeper_kafka_kafka_zooeeper_Windows_zookeeper_ _1_Windows 10_zookeeper_Zhang Phil-CSDN_Windows 10_zookeeper_1_zookeeper_https_archive.apache.org_dist_favicon32.ico_https_blog.csdn.net_zhangphil_article_details_123086790]: https://blog.csdn.net/zhangphil/article/details/123086790