docker安装kafka以及Python脚本生产消费


docker安装kafka以及Python脚本生产消费

1、kafka需要zookeeper管理,所以需要先安装zookeeper。

docker pull zookeeper:3.6

1.1、启动zookeeper

docker run -d --name zookeeper --publish 2181:2181 --volume /etc/localtime:/etc/localtime zookeeper:3.6

2、拉取kafka镜像

docker pull wurstmeister/kafka:2.12-2.5.0

2.1、启动kafka

docker run -d --name kafka --publish 9092:9092 --link zookeeper --env KAFKA_ZOOKEEPER_CONNECT=172.25.38.172:2181 --env KAFKA_ADVERTISED_HOST_NAME=172.25.38.172 --env KAFKA_ADVERTISED_PORT=9092 --volume /etc/localtime:/etc/localtime wurstmeister/kafka:2.12-2.5.0

2.2、备注:

KAFKA_ZOOKEEPER_CONNECT 配置本地的 ES 的地址,KAFKA_ADVERTISED_HOST_NAME 配置 kafka 所在服务器的地址,这里我都是本机 IP(需要把IP地址换成你的IP地址进行调试)

3、生产者脚本

# 生产者
import datetime
import json

from kafka import KafkaProducer

if __name__ == '__main__':

    topic = 'test'
    producer = KafkaProducer(bootstrap_servers='172.25.38.172:9092',
                             value_serializer=lambda m: json.dumps(m).encode("utf-8"))  # 连接kafka
    # 参数bootstrap_servers:指定kafka连接地址
    # 参数value_serializer:指定序列化的方式,我们定义json来序列化数据,当字典传入kafka时自动转换成bytes
    # 用户密码登入参数
    # security_protocol="SASL_PLAINTEXT"
    # sasl_mechanism="PLAIN"
    # sasl_plain_username="maple"
    # sasl_plain_password="maple"
    for i in range(100):
        data = {"num": i, "ts": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")}
        producer.send(topic, data)
    producer.close()

4、消费者脚本

import json

from kafka import KafkaConsumer

if __name__ == '__main__':

    consumer = KafkaConsumer("test",
                             bootstrap_servers="172.25.38.172:9092",
                             group_id='test13', )
    data = []
    for msg in consumer:
        info = msg.value.decode("utf-8")
        data = json.loads(info)
        print(data)

参考:

https://juejin.cn/post/6960820341631352868

https://www.icode9.com/content-4-1040800.html