python消息中间件 kafka

清疚 2022-01-05 13:25 353阅读 0赞

下载与安装 `Kafka`

wget http://apache.fayea.com/kafka/2.2.1/kafka_2.12-2.2.1.tgz
    tar -xzvf kafka_2.12-2.2.1.tgz 
    cd kafka_2.12-2.2.1
    
    # 先启动zookeeper（kafka依赖）
    bin/zookeeper-server-start.sh config/zookeeper.properties
    # 启动失败 @1 缺少jdk
    
    # 启动kafka
    bin/kafka-server-start.sh config/server.properties

`@1 安装JDK8`

# 首先下载，需要登录
    https://www.oracle.com/technetwork/java/javase/downloads/jdk8-downloads-2133151.html
    tar -xzvf jdk-8u211-linux-x64.tar.gz
    mv jdk1.8.0_211 /opt/jdk1.8
    
    
    vim /etc/profile
    
    ## JAVA
    
    export JAVA_HOME=/opt/jdk1.8
    export PATH=$PATH:$JAVA_HOME/bin
    #:wq 保存退出
    
    source /etc/profile
    # 验证
    java -version

**发送与接受消息**

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test1
    
    # --创建一个 名字 test1
    # --副本数量为 1
    # --partitions数量为1 的 topic
    
    Created topic test1. # 创建成功
    
    # 查看当前kafka中 存在哪些topic
    bin/kafka-topics.sh --list --zookeeper localhost:2181
    
    # 一般不需要创建topic，插入数据后，如果没有，则会自动创建
    
    # 生产者
    bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test1
    # 消费者
    bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test1 --from-beginning

`kafka python` 开发

# pip install kafka-python
    
    from kafka import KafkaProducer
    import names
    
    producer = KafkaProducer()
    
    for _ in range(10):
        name = names.get_full_name()
        future = producer.send('test',bytes(name,'ascii'))
        result = future.get(60)
        print(result)

from kafka import KafkaConsumer
    
    consumer = KafkaConsumer('test',group_id='test01')
    
    for msg in consumer:
        print(msg)