# Kafka

**Repository Path**: naclnezn/kafka

## Basic Information

- **Project Name**: Kafka
- **Description**: Kafka
- **Primary Language**: Unknown
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2024-08-10
- **Last Updated**: 2024-08-10

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# kafka

## 拉取(zk方法)

```shell
docker pull bitnami/kafka
```

## 运行

```shell
docker run -d --name kafka -p 9092:9092 --restart=always \
-e KAFKA_BROKER_ID=0 \
-e ALLOW_PLAINTEXT_LISTENER=yes \
-e KAFKA_CFG_ZOOKEEPER_CONNECT=192.168.0.234:2181 \
-e KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://192.168.0.234:9092 \
-e TZ="Asia/Shanghai" \
-v /tools/docker/kafka/data:/bitnami/kafka/data \
bitnami/kafka:latest
```

-e：指定环境变量
ALLOW_PLAINTEXT_LISTENER：允许使用PLAINTEXT侦听器
KAFKA_BROKER_ID=0：集群的broker的ID，该ID是集群的唯一标识
KAFKA_CFG_ZOOKEEPER_CONNECT：zk的连接地址
KAFKA_CFG_ADVERTISED_LISTENERS：kafka发布到zookeeper供客户端使用的服务地址(可被外部访问的地址)
TZ：指定时区，这里指定亚洲上海时区，也就是东八区

## 拉取(Raft方法)

```shell
docker pull bitnami/kafka:latest
```

```shell
kafka:
    image: registry.cn-hangzhou.aliyuncs.com/nacl-public/kafka:latest
    container_name: kafka
    user: root
    mem_limit: 512M # 限制内存为 512M
    ports:
      - 9092:9092
      - 9093:9093
    environment:
      - env TZ=Asia/Shanghai
      # 启用 Kafka Raft 模式
      - KAFKA_ENABLE_KRAFT=yes
      # 默认分区数
      - KAFKA_CFG_NUM_PARTITIONS=1
      # 控制台打印日志
      - BITNAMI_DEBUG=true
      # 设置 Kafka 的角色为 broker 和 controller
      - KAFKA_CFG_PROCESS_ROLES=broker,controller
      # 设置控制器监听器的名称
      - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER
      # 设置监听器的地址和端口
      - KAFKA_CFG_LISTENERS=SASL_PLAINTEXT://:9092,CONTROLLER://:9093
      # 设置监听器的协议
      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,SASL_PLAINTEXT:SASL_PLAINTEXT
      # 设置发布监听器的地址和端口
      - KAFKA_CFG_ADVERTISED_LISTENERS=SASL_PLAINTEXT://192.168.0.234:9092
      # 设置 broker 的 ID
      - KAFKA_BROKER_ID=1
      # 设置 节点 的 ID
      - KAFKA_CFG_NODE_ID=1
      # 设置集群 ID。你需要生成一个唯一的集群 ID 手写也行
      - KAFKA_KRAFT_CLUSTER_ID=your-cluster-id
      # 指定控制器仲裁投票者
      - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=1@192.168.0.234:9093
      # 允许使用 PLAINTEXT 监听器
      - ALLOW_PLAINTEXT_LISTENER=yes
      # 开启认证
      - KAFKA_OPTS=-Djava.security.auth.login.config=/opt/bitnami/kafka/conf/kafka_server_jaas.conf
      # 用户名
      - KAFKA_CLIENT_USERS=root
      # 密码
      - KAFKA_CLIENT_PASSWORDS=root
      # 设置 inter-broker 监听器名称
      - KAFKA_CFG_INTER_BROKER_LISTENER_NAME=SASL_PLAINTEXT
      # 设置 SASL 机制
      - KAFKA_CFG_SASL_MECHANISM_INTER_BROKER_PROTOCOL=PLAIN
    volumes:
      - /tools/docker/kafka/broker1:/bitnami/kafka:rw
      - /tools/docker/kafka/broker1/config/kafka_server_jaas.conf:/opt/bitnami/kafka/conf/kafka_server_jaas.conf
      - /tools/docker/kafka/broker1/config/client-sasl.properties:/opt/bitnami/kafka/conf/client-sasl.properties
    healthcheck: # 添加健康检测
      test:
        [
          "CMD",
          "kafka-broker-api-versions.sh",
          "--bootstrap-server",
          "192.168.0.234:9092",
          "--command-config",
          "/opt/bitnami/kafka/conf/client-sasl.properties"
        ]
      interval: 30s
      timeout: 3s
      retries: 3
      start_period: 25s
    depends_on:
      redis: # 在mysql启动后启动
        condition: service_healthy
    restart: unless-stopped
```

client-sasl.properties

```properties
sasl.mechanism=PLAIN
security.protocol=SASL_PLAINTEXT
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="root" password="root";
```

kafka_server_jaas.conf

```config
KafkaServer {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="root"
    password="root"
    user_admin="root";
};
```