部署kafka

docker-compose

编写docker-compose.yml，通过docker容器部署单节点kafka

version: '3'
services:
    zookeeper: 
        image: wurstmeister/zookeeper:3.4.6 
        volumes: 
            - ./zookeeper_data:/opt/zookeeper-3.4.6/data 
        container_name: zookeeper 
        ports: 
            - "10002:2181" 
            - "10003:2182" 
        restart: always
​
    kafka: 
        image: wurstmeister/kafka 
        container_name: kafka_01 
        depends_on: 
            - zookeeper 
        ports: 
            - "10004:9092" 
        volumes: 
            - ./kafka_log:/kafka 
        environment: 
            - KAFKA_BROKER_NO=0 
            - KAFKA_BROKER_ID=0 
            - KAFKA_LISTENERS=PLAINTEXT://kafka_01:9092                     # kafka tcp 侦听的ip
            - KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://服务器ip:10004        # kafka broker侦听的ip
            - KAFKA_INTER_BROKER_LISTENER_NAME=PLAINTEXT 
            - KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181 
            - KAFKA_HEAP_OPTS=-Xmx512M -Xms16M
        restart: always
    # kafka集群管理面板
    kafka_manager: 
        image: sheepkiller/kafka-manager 
        ports: 
            - "10005:9000" 
        environment: 
            - ZK_HOSTS=zookeeper:2181 
        depends_on: 
            - zookeeper 
            - kafka 
        restart: always

 

后台运行

docker-compose up -d

 

docker ps命令查看容器是否启动成功

通过上述docker-compose.yml部署会运行三个容器，选择进入kafka容器

docker exec -it kafka容器id /bin/bash
# 进入kafka目录
cd /opt/kafka_2.13-2.8.1/

在容器内创建topic，topic是kafka中数据管理的基本单位，或者说集合，每一个topic可以管理多个partition，编码操作时：你可以往对应kafka服务器ip+port+topic+partition去发送和读取数据。

bin/kafka-topics.sh --create --zookeeper 服务器ip:2181 --replication-factor 1 -partitions 1 --topic test

业务编写

Go语言中连接kafka使用第三方库: github.com/Shopify/sarama

go get github.com/segmentio/kafka-go

sarama库的简易操作可以参照文档（消费者的编写文档中有坑）：文档地址

如下使用kafka client库进行编码所涉及的API操作比较简单，流程上或许不够规范，请酌情参考。

producer

文档中生产者只发送了一条数据后就会关闭，这里我改成了每秒钟发送一次。

consumer

文档中消费者虽然开启了Go协程（类比于Java的线程）去读取kafka的数据，但是由于主程序执行顺序执行完毕后，子协程也会终止，导致子协程还没有读取成功/打印数据，整个程序就已经关闭运行了。

因此我做了一些改动，在子协程退出之前，保持主程序不会退出（使用Go语言的WaitGroup），如果简单粗暴在main函数末尾设置一个很长的程序sleep时间，也是可以实现打印输出的。

生产&消费

确保kafka容器正常运行，kafka服务器防火墙端口正常开放，运行消费者程序，运行生产者程序。这个生产者每秒向kafka发送一条测试数据：this is a test log，你也可以添加上程序运行时间进行测试。

事实上被客户端消费后的数据并没有马上从kafka删除，这里不多做介绍，各位自行了解～

小结

本文讲解了使用docker-compose部署单节点kafka的流程，后续通过修改docker-compose.yml的内容也可以实现kafka集群的部署，并且，在较新版本的kafka中，集群的部署可以脱离zookeeper，但是经过了解，由于功能并不完善，这里还是选择了基于zookeeper的部署。

 

转载至：https://www.cnblogs.com/YLTFY1998/p/16550406.html

最后修改于
上一篇