创建Docker-kafka-cluster目录，编写docker-compose.yml文件

docker-compose.yml

# Copyright Broadcom, Inc. All Rights Reserved.
# SPDX-License-Identifier: APACHE-2.0

services:
    kafka-0:
      container_name: kafka-0
      hostname: kafka-0
      image: docker.io/bitnami/kafka:3.9
    #   ports:
    #     - "9092"
      environment:
        # KRaft settings
        - KAFKA_CFG_NODE_ID=0
        - KAFKA_CFG_PROCESS_ROLES=controller,broker
        - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=0@kafka-0:9093,1@kafka-1:9093,2@kafka-2:9093
        - KAFKA_KRAFT_CLUSTER_ID=abcdefghijklmnopqrstuv
        # Listeners
        - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093,EXTERNAL://0.0.0.0:9094
        - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://kafka-0:9092,EXTERNAL://localhost:9094
        - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT,PLAINTEXT:PLAINTEXT
        - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER
        - KAFKA_CFG_INTER_BROKER_LISTENER_NAME=PLAINTEXT
        # Clustering
        - KAFKA_CFG_OFFSETS_TOPIC_REPLICATION_FACTOR=3
        - KAFKA_CFG_TRANSACTION_STATE_LOG_REPLICATION_FACTOR=3
        - KAFKA_CFG_TRANSACTION_STATE_LOG_MIN_ISR=2
      volumes:
        - kafka_0_data:/bitnami/kafka
    kafka-1:
      container_name: kafka-1
      hostname: kafka-1
      image: docker.io/bitnami/kafka:3.9
    #   ports:
    #     - "9092"
      environment:
        # KRaft settings
        - KAFKA_CFG_NODE_ID=1
        - KAFKA_CFG_PROCESS_ROLES=controller,broker
        - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=0@kafka-0:9093,1@kafka-1:9093,2@kafka-2:9093
        - KAFKA_KRAFT_CLUSTER_ID=abcdefghijklmnopqrstuv
        # Listeners
        - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093,EXTERNAL://0.0.0.0:9095
        - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://kafka-1:9092,EXTERNAL://localhost:9095
        - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT,PLAINTEXT:PLAINTEXT
        - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER
        - KAFKA_CFG_INTER_BROKER_LISTENER_NAME=PLAINTEXT
        # Clustering
        - KAFKA_CFG_OFFSETS_TOPIC_REPLICATION_FACTOR=3
        - KAFKA_CFG_TRANSACTION_STATE_LOG_REPLICATION_FACTOR=3
        - KAFKA_CFG_TRANSACTION_STATE_LOG_MIN_ISR=2
      volumes:
        - kafka_1_data:/bitnami/kafka
    kafka-2:
      container_name: kafka-2
      hostname: kafka-2
      image: docker.io/bitnami/kafka:3.9
    #   ports:
    #     - "9092"
      environment:
        # KRaft settings
        - KAFKA_CFG_NODE_ID=2
        - KAFKA_CFG_PROCESS_ROLES=controller,broker
        - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=0@kafka-0:9093,1@kafka-1:9093,2@kafka-2:9093
        - KAFKA_KRAFT_CLUSTER_ID=abcdefghijklmnopqrstuv
        # Listeners
        - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093,EXTERNAL://0.0.0.0:9096
        - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://kafka-2:9092,EXTERNAL://localhost:9096
        - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT,PLAINTEXT:PLAINTEXT
        - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER
        - KAFKA_CFG_INTER_BROKER_LISTENER_NAME=PLAINTEXT
        # Clustering
        - KAFKA_CFG_OFFSETS_TOPIC_REPLICATION_FACTOR=3
        - KAFKA_CFG_TRANSACTION_STATE_LOG_REPLICATION_FACTOR=3
        - KAFKA_CFG_TRANSACTION_STATE_LOG_MIN_ISR=2
      volumes:
        - kafka_2_data:/bitnami/kafka
     
    nginx:
        container_name: nginx
        hostname: nginx
        image: nginx:1.22.0-alpine
        volumes:
        - ./nginx.conf:/etc/nginx/nginx.conf:ro
        ports:
            - "9094-9096:9094-9096"
        depends_on:
            - kafka-0
            - kafka-1
            - kafka-2

        
volumes:
    kafka_0_data:
      driver: local
    kafka_1_data:
      driver: local
    kafka_2_data:
      driver: local

编写nginx.conf配置文件，实现负载均衡

nginx.conf

user nginx;
worker_processes auto;

error_log /var/log/nginx/error.log notice;
pid       /var/run/nginx.pid;

events{
    worker_connections 1024;
}

stream{
    upstream kafka-0 {
        server kafka-0:9094;
    }
    upstream kafka-1 {
        server kafka-1:9095;
    }
    upstream kafka-2 {
        server kafka-2:9096;
    }
    server {
        listen 9094;
        proxy_pass kafka-0;
    }
    server {
        listen 9095;
        proxy_pass kafka-1;
    }
    server {
        listen 9096;
        proxy_pass kafka-2;
    }

}

http {
    include        /etc/nginx/mime.types;
    default_type    application/octet-stream;
    log_format main     'remote_addr - $remote_user [$time_local] "$request" '
                        '$status $body_bytes_sent "$http_referer" '
                        '"$http_user_agent" "$http_x_forwarded_for"';
    access_log /var/log/nginx/access.log    main;
    
    sendfile        on;
    #tcp_nopush     on;

    keepalive_timeout   65;
    #gzip           on;
    include /etc/nginx/conf.d/*.conf;
}

启动部署

docker compose up -d

创建主题

docker exec -it kafka-0 /opt/bitnami/kafka/bin/kafka-topics.sh \
--create --bootstrap-server kafka-0:9092 \
--topic my-topic \
--partitions 3 --replication-factor 2

控制台生产者

docker exec -it kafka-0 /opt/bitnami/kafka/bin/kafka-console-producer.sh \
--bootstrap-server kafka-0:9092 \
--topic my-topic

控制台消费者

docker exec -it kafka-0 /opt/bitnami/kafka/bin/kafka-console-consumer.sh \
--bootstrap-server kafka-0:9092 \
--topic my-topic

至多一次

//生产者
acks=0 or 1
retries=0

//消费者
props.put("enable.auto.commit, "true");
props.put("auto.commit.interval.ms", "1000");
//todo
consumer.poll(100);

至少一次

//生产者
acks=all or -1
retries>0

//消费者
props.put("enable.auto.commit, "false");
//todo
consumer.commitSync();

精确一次

//生产者
enable.idempotence=true
retries=Integer.MAX_VALUE
acks=all or -1

//消费者
业务唯一id