如何在私有容器中配置Kafka生产者并向Kafka消费者推送消息？

核心思路 #

你需要将原Compose中的3个Kafka客户端容器替换为实际业务服务（如CouchDB、Vitess Orchestrator），同时让这些服务能够与Kafka集群交互（创建Topic、推送消息）。核心要点是保证服务能访问Kafka集群，并通过合适的方式完成消息发送（这类服务本身不带Kafka客户端）。

修改后的Docker Compose配置示例 #

version: '3'
services:
  zookeeper:
    image: 'confluentinc/cp-zookeeper:latest'
    container_name: zookeeper
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
    healthcheck:
      test: ["CMD", "zkServer.sh", "status"]
      interval: 10s
      timeout: 5s
      retries: 3

  kafka:
    image: 'confluentinc/cp-kafka:latest'
    container_name: kafka-server
    depends_on:
      zookeeper:
        condition: service_healthy
    environment:
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_LISTENERS: PLAINTEXT://:9092
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_BROKER_ID: 1
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'true' # 自动创建不存在的Topic，生产环境可关闭
    healthcheck:
      test: ["CMD", "kafka-topics.sh", "--list", "--bootstrap-server", "localhost:9092"]
      interval: 10s
      timeout: 5s
      retries: 3

  # 替换为非Kafka服务示例1：CouchDB
  couchdb:
    image: couchdb:latest
    container_name: couchdb-service
    depends_on:
      kafka:
        condition: service_healthy
    ports:
      - "5984:5984"
    environment:
      COUCHDB_USER: admin
      COUCHDB_PASSWORD: admin

  # 替换为非Kafka服务示例2：Vitess Orchestrator
  vitess-orchestrator:
    image: vitess/orchestrator:latest
    container_name: vitess-orchestrator-service
    depends_on:
      kafka:
        condition: service_healthy
    ports:
      - "3000:3000"
    command: ["orchestrator", "-config", "/etc/orchestrator/config.json"]
    volumes:
      - ./orchestrator-config.json:/etc/orchestrator/config.json # 挂载自定义配置

  # 替换为非Kafka服务示例3：自定义业务服务（示例）
  custom-service:
    image: your-custom-image:latest
    container_name: custom-business-service
    depends_on:
      kafka:
        condition: service_healthy
    environment:
      KAFKA_BROKER_URL: kafka:9092 # 注入Kafka地址到环境变量

关键配置说明 #

1. 网络访问：Docker Compose默认会为所有服务创建一个共享网络，因此非Kafka服务可以直接通过服务名kafka:9092访问Kafka集群，无需额外配置端口映射（除非需要从宿主机访问）。
2. 健康检查：给Zookeeper和Kafka添加健康检查，确保它们完全就绪后再启动业务服务，避免服务启动时Kafka尚未可用的问题。
3. 环境变量注入：将Kafka地址通过环境变量（如KAFKA_BROKER_URL）传递给业务服务，方便服务内部调用。

非Kafka服务推送消息到Kafka的实现方式 #

由于这些服务本身不带Kafka客户端，你可以选择以下几种方案：

方案1：使用Kafka REST代理（推荐新手）

部署一个Kafka REST代理服务，让业务服务通过HTTP请求与Kafka交互，无需安装任何Kafka客户端。在Compose中添加以下服务：

kafka-rest:
    image: confluentinc/cp-kafka-rest:latest
    container_name: kafka-rest-proxy
    depends_on:
      kafka:
        condition: service_healthy
    environment:
      KAFKA_REST_BOOTSTRAP_SERVERS: PLAINTEXT://kafka:9092
      KAFKA_REST_LISTENERS: http://0.0.0.0:8082
    ports:
      - "8082:8082"

业务服务可以通过以下HTTP请求操作Kafka：

• 创建Topic：POST http://kafka-rest:8082/topics/my-topic，请求体：{"partitions":1,"replication_factor":1}
• 发送消息：POST http://kafka-rest:8082/topics/my-topic，请求体：{"records":[{"value":"hello world"}]}

方案2：自定义镜像安装Kafka客户端

如果业务服务需要直接使用Kafka命令行工具，可以基于官方镜像自定义Dockerfile，安装Kafka客户端。例如CouchDB的Dockerfile：

FROM couchdb:latest
RUN apt-get update && apt-get install -y openjdk-17-jre-headless wget
RUN wget https://downloads.apache.org/kafka/3.6.1/kafka_2.13-3.6.1.tgz && \
    tar xzf kafka_2.13-3.6.1.tgz && \
    mv kafka_2.13-3.6.1 /opt/kafka
ENV PATH="/opt/kafka/bin:${PATH}"

构建镜像后替换原CouchDB镜像即可，服务内部可以直接使用kafka-console-producer.sh等工具发送消息。

方案3：利用服务自身集成能力

部分服务（如CouchDB）支持通过Webhook或插件触发外部请求，你可以配置这类服务在特定事件（如文档创建）发生时，调用Kafka REST代理发送消息。

Topic创建方式 #

• 自动创建：通过Kafka的KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'true'配置，当生产者发送消息到不存在的Topic时，Kafka会自动创建（生产环境建议关闭，手动管理Topic）。
• 手动创建：可以通过临时运行Kafka客户端容器执行命令：

docker run --rm --network <your-compose-network> confluentinc/cp-kafka:latest kafka-topics.sh \
  --create --topic my-topic --bootstrap-server kafka:9092 \
  --partitions 1 --replication-factor 1

（替换<your-compose-network>为你的Compose默认网络名，可通过docker network ls查看）

内容的提问来源于stack exchange，提问作者Avinash