kafka消息队列的使用

概念性内容参考 Kafka的简介

软件的安装参考 mac 安装Kafka

python代码使用 ，先 pip3 install kafka-python

生产者

from kafka import KafkaProducer

# 生产端
producer = KafkaProducer(bootstrap_servers=[\'localhost:9092\'])
for i in range(3):
    future = producer.send(
              \'test\',   # 设置topic，消费者根据topic拿数据
              key=b\'qwedef\',    # 必须是bytes数据类型，同一个key值，会被送至同一个分区
              value=b\'dewcdef\',     # 必须是bytes数据类型， 发送的数据
              partition=0      # partition设置分区
    )  
    result = future.get(timeout=10)      # 监控是否发送成功         
    # print(result)

消费者

from kafka import KafkaConsumer

# 消费端

consumer = KafkaConsumer(
                \'test\',        # 设置topic，从topic里面拿数据
                group_id=\'group1\',    # 如果不设置这个参数，则消费者A消费后，消费者B会继续消费；如果和其他消费者设置同样的group_id， 则消费者A消费这条数据后，不会被消费者B消费
                bootstrap_servers=[\'localhost:9092\']            
)
for msg in consumer:
    print(msg.key.decode())
    print(msg.value.decode())