Docker Compose部署Kafka集群并在宿主机Windows连接开发

🕗 发布于 2024-07-15 05:26 docker kafka windows

文章目录

1. 常用参数
2. 理解参数和原理
3. Docker Compose
4. 验证

Docker for Windows	4.23.0
windows	11
Java	17

1. 常用参数

kafka容器常用参数如下

-e KAFKA_BROKER_ID=1：设置 Kafka broker 的 ID 为 1。每个 Kafka broker 都需要一个唯一的 ID。
-e KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181：指定 Kafka 连接到 Zookeeper 的地址，这里假设 Zookeeper 容器的名称为 zookeeper，并且它在 2181 端口监听。
-e ALLOW_PLAINTEXT_LISTENER=yes：允许 Kafka 使用纯文本监听器。即允许非加密的通信。
-e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092：Kafka broker 实际监听在容器内的 0.0.0.0:9092 上。这意味着 Kafka 接受来自任何网络接口的连接。
-e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092：指定 Kafka 广播其监听器地址，客户端将使用该地址连接到 broker。在这个例子中，Kafka 广播它在 localhost:9092 上监听。
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP：指定 Kafka 使用的监听器协议映射。例如：PLAINTEXT:PLAINTEXT,SSL:SSL。
KAFKA_INTER_BROKER_LISTENER_NAME：指定 broker 间通信使用的监听器名称。例如：PLAINTEXT。

2. 理解参数和原理

KAFKA_LISTENERS是broker实际监听的地址。

KAFKA_ADVERTISED_LISTENERS是broker注册在zookeeper或者controller broker里面的元数据，当消费者或者生产者使用Bootstrap-Server去连接kafka集群时，集群会返回元数据等信息到客户端，客户端会根据每个broker提供的KAFKA_ADVERTISED_LISTENERS去连接对应的broker。

所以首先，集群之间，broker之间需要通信，所以每个kafka容器需要设置一个KAFKA_ADVERTISED_LISTENERS用于告诉别的容器如何连接到自己，如果容器都是处于同一bridge网络，那么直接使用容器名即可。

其次，我们想要在宿主机比如windows的idea开发，我们一般只能通过docker容器-p暴露的端口去连接kafka，所以每个kafka容器还需要设置一个KAFKA_ADVERTISED_LISTENERS来告诉宿主机的客户端，如何连接到自己，这里需要使用localhost+暴露在宿主机的端口。

那么如果KAFKA_ADVERTISED_LISTENERS里面有2个地址，如何保证broker之间的连接使用的是容器名，而宿主机客户端使用的是localhost呢？

这需要KAFKA_INTER_BROKER_LISTENER_NAME来指定前者。

并且由于KAFKA_ADVERTISED_LISTENERS里面有2个地址，所以我们还需要KAFKA_LISTENER_SECURITY_PROTOCOL_MAP来映射监听器名字。

3. Docker Compose

version: '3.8'

services:
  zookeeper:
    image: bitnami/zookeeper:3.8.2
    container_name: zookeeper
    environment:
      - ALLOW_ANONYMOUS_LOGIN=yes
    networks:
      - kafka

  kafka1:
    image: bitnami/kafka:3.6.1
    container_name: kafka1
    depends_on:
      - zookeeper
    ports:
      - "19092:9092"
    environment:
      - KAFKA_BROKER_ID=1
      - KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181
      - KAFKA_LISTENERS=INTERNAL://0.0.0.0:9093,EXTERNAL://0.0.0.0:9092
      - KAFKA_ADVERTISED_LISTENERS=INTERNAL://kafka1:9093,EXTERNAL://localhost:19092
      - KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT
      - KAFKA_INTER_BROKER_LISTENER_NAME=INTERNAL
    networks:
      - kafka

  kafka2:
    image: bitnami/kafka:3.6.1
    container_name: kafka2
    depends_on:
      - zookeeper
    ports:
      - "29092:9092"
    environment:
      - KAFKA_BROKER_ID=2
      - KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181
      - KAFKA_LISTENERS=INTERNAL://0.0.0.0:9093,EXTERNAL://0.0.0.0:9092
      - KAFKA_ADVERTISED_LISTENERS=INTERNAL://kafka2:9093,EXTERNAL://localhost:29092
      - KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT
      - KAFKA_INTER_BROKER_LISTENER_NAME=INTERNAL
    networks:
      - kafka

  kafka3:
    image: bitnami/kafka:3.6.1
    container_name: kafka3
    depends_on:
      - zookeeper
    ports:
      - "39092:9092"
    environment:
      - KAFKA_BROKER_ID=3
      - KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181
      - KAFKA_LISTENERS=INTERNAL://0.0.0.0:9093,EXTERNAL://0.0.0.0:9092
      - KAFKA_ADVERTISED_LISTENERS=INTERNAL://kafka3:9093,EXTERNAL://localhost:39092
      - KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT
      - KAFKA_INTER_BROKER_LISTENER_NAME=INTERNAL
    networks:
      - kafka

networks:
  kafka:
    driver: bridge

可以看到每个容器都设置了INTERNAL，因为指定了KAFKA_INTER_BROKER_LISTENER_NAME=INTERNAL，所以这是用于broker之间的连接，其监听在本地的0.0.0.0:9093，广播给其它broker的通信地址是<容器名>:9093，使用PLAINTEXT（明文）方式通信。

除此之外还设置了EXTERNAL，监听在本地的0.0.0.0:9092，广播给客户端的地址是localhost:19092、localhost:29092、localhost:39092，也就是windows上的客户端通过localhost:19092访问broker，这会被docker的-p映射到对应容器的9092，被0.0.0.0:9092对接。

4. 验证

连接到某个容器。创建test主题。

kafka-topics.sh --create --topic test --partitions 3 --replication-factor 3 --bootstrap-server kafka1:9093

查看分区和副本情况，可以看到在不同的broker上，输出中显示的是Broker ID。

I have no name!@7212060b6e3d:/$ kafka-topics.sh --describe --topic test --bootstrap-server kafka1:9093
Topic: test     TopicId: Lo1eQ6aCQj6WiFcNiVBrcw PartitionCount: 3       ReplicationFactor: 3    Configs: 
        Topic: test     Partition: 0    Leader: 2       Replicas: 2,3,1 Isr: 2,3,1
        Topic: test     Partition: 1    Leader: 3       Replicas: 3,1,2 Isr: 3,1,2
        Topic: test     Partition: 2    Leader: 1       Replicas: 1,2,3 Isr: 1,2,3

引入pom包

<dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>3.6.1</version>
        </dependency>
        <!-- SLF4J API -->
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
            <version>1.7.36</version>
        </dependency>
        <!-- Logback classic (SLF4J implementation) -->
        <dependency>
            <groupId>ch.qos.logback</groupId>
            <artifactId>logback-classic</artifactId>
            <version>1.2.11</version>
        </dependency>

生产者代码，通过localhost:19092连接到集群。

package org.dragon.producer;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.HashMap;

public class KafkaProducerTest {
    public static void main(String[] args) throws InterruptedException {
        //创建producer
        HashMap<String, Object> config = new HashMap<>();
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:19092");
        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(config);

        for (int i = 0; i < 10; i++) {
            //创建record
            ProducerRecord<String, String> record = new ProducerRecord<String, String>(
                    "test",
                    "key"+i,
                    "我是你爹"+i
            );
            //发送record
            producer.send(record);
            Thread.sleep(500);
        }

        //关闭producer
        producer.close();
    }
}

消费者代码，

package org.dragon.consumer;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Collections;
import java.util.HashMap;

public class KafkaConsumerTest {
    public static void main(String[] args) {

        // 创建消费者对象
        HashMap<String, Object> config = new HashMap<>();
        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:19092");
        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        config.put(ConsumerConfig.GROUP_ID_CONFIG, "mkl");

        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(config);
        // 消费者订阅主题
        kafkaConsumer.subscribe(Collections.singletonList("test"));

        try {
            while (true){
                // 消费者拉取消息
                ConsumerRecords<String, String> records = kafkaConsumer.poll(100);
                records.forEach(System.out::println);
            }
        }finally {
            // 消费者关闭
            kafkaConsumer.close();
        }
    }
}

都启动后，消费者和生产者日志正常。
在这里插入图片描述

原文地址：https://blog.csdn.net/m0_51390969/article/details/140422150

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

上一篇：Github 2024-07-14 php开源项目日报 Top10
下一篇：css文字自适应宽度动态出现省略号...

基于RK3588的移动充电机器人应用解决方案
伴随着国内新能源汽车保有量的持续增长，充电难题愈发凸显，配套充电设施的建设需求进一步扩大。可外接4G/5G、WiFi模块，用于传感数据、视频数据的上传，同时，通过网络的连接，能够远程监控机器人的状态和
阅读更多2024-11-16
开源，一天200star，解锁视频字幕生成新方式——一款轻量级开源字幕工具，免费，支持花字，剪映最新会员模式吃相太难看了
是一款基于开源的轻量级视频字幕生成工具，由开发者 @WEIFENG2333 精心打造。其主要目的是为用户提供一套免费、易用的字幕生成方案，支持各种视频格式，并结合现代化的自然语言处理技术，实现精准、高
阅读更多2024-11-16
docker：基于Dockerfile镜像制作完整案例
docker：基于Dockerfile镜像制作完整案例
阅读更多2024-11-16
【数据库系列】 Spring Boot 集成 Neo4j 的详细介绍
Spring Boot 提供了对 Neo4j 的良好支持，使得开发者可以更方便地使用图数据库。通过使用 Spring Data Neo4j，开发者可以轻松地进行数据访问、操作以及管理。本文将详细介绍如
阅读更多2024-11-16
微信小程序02-页面制作
微信小程序提供了丰富的组件，如view、image等，用于构建具有微信风格的UI界面。视口单位用于移动端页面适配，如vw和vh。表单组件用于收集用户信息，如form、button、input等。
阅读更多2024-11-16
2024 CCF中国开源大会“开源科学计算与系统建模openSCS”分论坛成功举办
目前MWORKS用户已遍及国内外295所高校，仅深圳就有41家应用验证企业，覆盖新能源、芯片、车辆、低空经济、医疗、制造、自动化、电力、航天、金融、船舶等优势行业单位，已经具备良好的生态基础。该委员会
阅读更多2024-11-16
C&C++内存管理
pChar3是指针变量，存在栈中，*pChar3是由const修饰的，pChar3指向的内容不能修改，内容具有常属性，因此pChae3存在代码段中，也就是常量区。内存泄漏并不是指内存在物理上的消失，而
阅读更多2024-11-16
单片机面试
答案：宏定义是在编译预处理阶段被处理的。预处理包含：头文件包含，宏替换，条件编译，去除注释，添加行号。
阅读更多2024-11-16
Ubuntu23.10下解决C语言调用mysql.h问题
在学习C语言和MySQL的调用的时候遇到包和版本的问题，由于使用的书很老（10年的），因此很多MySQL的包已经过时，在查找很多资料和询问gpt之后得到了解决方案
阅读更多2024-11-16
高级java每日一道面试题-2024年11月07日-Redis篇-Redis有哪些功能?
Redis 是一个功能强大且灵活的 NoSQL 数据库，广泛应用于缓存、消息队列、实时分析等多种场景。在高级 Java 面试中，了解 Redis 的数据类型、持久化、发布/订阅、事务、Lua 脚本、主
阅读更多2024-11-16