用 Docker 快速搭建 Kafka 集群

忘是亡心i 2021-07-26 12:38 486阅读 0赞

![1803089889.png][]

开源Linux

一个执着于技术的公众号

![undefined][]

![format_png][]

### 版本    ###

•JDK 14•Zookeeper•Kafka

### 安装 Zookeeper 和 Kafka ###

Kafka 依赖 Zookeeper，所以我们需要在安装 Kafka 之前先拥有 Zookeeper。准备如下的 docker-compose.yaml 文件，将文件中的主机地址 192.168.1.100 替换成你自己的环境中的主机地址即可。

version: "3"
    services:
      zookeeper:
        image: zookeeper
        build:
          context: ./
        container_name: zookeeper
        ports:
          - 2181:2181
        volumes:
          - ./data/zookeeper/data:/data
          - ./data/zookeeper/datalog:/datalog
          - ./data/zookeeper/logs:/logs
        restart: always
      kafka_node_0:
        depends_on:
          - zookeeper
        build:
          context: ./
        container_name: kafka-node-0
        image: wurstmeister/kafka
        environment:
          KAFKA_BROKER_ID: 0
          KAFKA_ZOOKEEPER_CONNECT: 192.168.1.100:2181
          KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://192.168.1.100:9092
          KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092
          KAFKA_NUM_PARTITIONS: 3
          KAFKA_DEFAULT_REPLICATION_FACTOR: 2
        ports:
          - 9092:9092
        volumes:
          - ./data/kafka/node_0:/kafka
        restart: unless-stopped
      kafka_node_1:
        depends_on:
          - kafka_node_0
        build:
          context: ./
        container_name: kafka-node-1
        image: wurstmeister/kafka
        environment:
          KAFKA_BROKER_ID: 1
          KAFKA_ZOOKEEPER_CONNECT: 192.168.1.100:2181
          KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://192.168.1.100:9093
          KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9093
          KAFKA_NUM_PARTITIONS: 3
          KAFKA_DEFAULT_REPLICATION_FACTOR: 2
        ports:
          - 9093:9093
        volumes:
          - ./data/kafka/node_1:/kafka
        restart: unless-stopped
      kafka_node_2:
        depends_on:
          - kafka_node_1
        build:
          context: ./
        container_name: kafka-node-2
        image: wurstmeister/kafka
        environment:
          KAFKA_BROKER_ID: 2
          KAFKA_ZOOKEEPER_CONNECT: 192.168.1.100:2181
          KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://192.168.1.100:9094
          KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9094
          KAFKA_NUM_PARTITIONS: 3
          KAFKA_DEFAULT_REPLICATION_FACTOR: 2
        ports:
          - 9094:9094
        volumes:
          - ./data/kafka/node_2:/kafka
        restart: unless-stopped

输入 docker-compose up -d 运行脚本文件进行集群构建。等待一会儿，得到如下结果即为成功。

![null][]

### SpringBoot 集成 Kafka 集群 ###

创建一个全新的 SpringBoot 工程，在 build.gradle 文件中添加下列依赖。

dependencies {
        ...
        ...
        implementation 'org.springframework.kafka:spring-kafka:2.5.2.RELEASE'
        implementation 'com.alibaba:fastjson:1.2.71'
    }

1.在 application.properties 进行 Kafka 相关参数配置。

spring.kafka.bootstrap-servers=192.168.1.100:9092,192.168.1.100:9093,192.168.1.100:9094
    spring.kafka.producer.retries=0
    spring.kafka.producer.batch-size=16384
    spring.kafka.producer.buffer-memory=33554432
    spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
    spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
    spring.kafka.consumer.auto-offset-reset=latest
    spring.kafka.consumer.enable-auto-commit=true
    spring.kafka.consumer.auto-commit-interval=100

2.创建消息体类。

public class Message {
        private Long id;
        private String message;
        private Date sendAt;
    }

3.创建消息发送者

public class Sender {
        @Autowired
        private KafkaTemplate<String, String> kafkaTemplate;
        public void send() {
            Message message = new Message();
            message.setId(System.currentTimeMillis());
            message.setMessage(UUID.randomUUID().toString());
            message.setSendAt(new Date());
            log.info("message = {}", JSON.toJSONString(message));
            kafkaTemplate.send("test", JSON.toJSONString(message));
        }
    }

4.创建消息接收者

public class Receiver {
        @KafkaListener(topics = {"test"}, groupId = "test")
        public void listen(ConsumerRecord<?, ?> record) {
            Optional<?> message = Optional.ofNullable(record.value());
            if (message.isPresent()) {
                log.info("receiver record = " + record);
                log.info("receiver message = " + message.get());
            }
        }
    }

5.测试消息队列

public class QueueController {
        @Autowired
        private Sender sender;
        @PostMapping("/test")
        public void testQueue() {
            sender.send();
            sender.send();
            sender.send();
        }
    }

得到如下日志即为集成成功。

![null][null 1]

到这里就我们就成功搭建了一个 Kafka 伪集群，并成功与 SpringBoot 进行整合。

如果你想了解Docker容器技术及应用，欢迎参加8月7日晚的公开课。本次公开课主要围绕Docker容器技术入门和应用实战，系统介绍Docker容器技术、在企业中应用、以及网络管理、储存卷管理和部署Nginx、PHP、MySQL、Redis实战等，干货满满。

作者：曾是然

链接：https://segmentfault.com/a/1190000022988499

**- End -**

**关注「开源Linux」加星标，提升IT技能**

![format_png 1][]

**好文章，****分享****、****点赞****、****在看****三连哦****❤️****↓↓↓**

[1803089889.png]: /images/20210726/b8a23fc1cfff4c60800c7d6781ff7d85.png
[undefined]: /images/20210726/f526c44b12f7468fa3e80ab5a65f4bb9.png
[format_png]: /images/20210726/a182bc8c9fb848599de192dd6c1ccca9.png
[null]: /images/20210726/d67e5961bf4f4517a1098140366c3b58.png
[null 1]: /images/20210726/cc7034d9c42e4fa7997ac674f1747c8a.png
[format_png 1]: /images/20210726/285658dd93ee4a16ab464776b2a37682.png