argoproj/kafka-gateway

Kafka Gateway 镜像文档

镜像概述

Kafka Gateway是一款轻量级中间件，作为Kafka集群的统一入口网关，旨在简化外部应用与Kafka的集成过程。通过提供标准化接入点，该镜像实现了协议转换、请求路由、负载均衡及安全管控等核心能力，有效降低直接访问Kafka集群的复杂性，同时增强系统的可管理性、安全性和扩展性。

核心功能与特性

核心功能

• 协议转换：支持HTTP/REST、MQTT、AMQP等多种协议与Kafka协议的双向转换，适配不同外部系统接入需求。
• 消息路由：基于主题、分区、消息元数据等维度实现灵活路由策略，支持消息过滤与转发。
• 负载均衡：智能分发请求至Kafka brokers，优化集群资源利用率，避免单点负载过高。
• 安全控制：集成认证（Basic Auth、OAuth2）、授权及TLS加密，保障Kafka集群访问安全。

关键特性

• 轻量级部署：容器化设计，资源占用低，支持快速启停与水平扩展。
• 监控集成：暴露Prometheus metrics接口，支持与Grafana等监控系统对接，实时监控网关运行状态。
• 高可用性：支持集群部署模式，通过主从切换或分布式架构确保服务连续性。
• 动态配置：支持配置热更新，无需重启即可调整路由规则、认证策略等参数。

使用场景与适用范围

典型使用场景

• 多协议接入：外部应用使用HTTP/REST或MQTT协议需与Kafka交互时（如IoT设备、Web应用）。
• 统一接入管控：多团队或多应用共享Kafka集群时，需隔离访问权限并统一流量管理。
• 安全增强：需对Kafka访问进行认证授权、数据加密，或隐藏集群拓扑信息时。
• 消息预处理：需对入站/出站消息进行格式转换、过滤或 enrichment 时（如JSON与Avro格式转换）。

适用范围

• 中小型Kafka集群的外部接入层建设
• 需跨协议集成Kafka的异构系统架构
• 对Kafka访问有严格安全与管控需求的场景
• 需快速扩展Kafka接入能力的业务场景

部署与使用方法

Docker Run 部署

通过docker run快速启动单节点Kafka Gateway：

bash
docker run -d \
  --name kafka-gateway \
  -p 8080:8080 \  # HTTP协议端口
  -p 1883:1883 \  # MQTT协议端口（如启用）
  -e KAFKA_BROKERS=kafka1:9092,kafka2:9092 \  # Kafka集群broker列表（必填）
  -e PROTOCOL_TYPE=HTTP,MQTT \  # 启用的协议类型，多协议用逗号分隔
  -e AUTH_ENABLED=true \  # 启用Basic Auth认证
  -e AUTH_USERNAME=admin \  # 认证用户名
  -e AUTH_PASSWORD=secure-pass \  # 认证密码
  -e TLS_ENABLED=true \  # 启用TLS加密
  -e TLS_CERT_PATH=/etc/certs/server.crt \  # 挂载的TLS证书路径
  -e TLS_KEY_PATH=/etc/certs/server.key \  # 挂载的TLS私钥路径
  -v ./local-certs:/etc/certs \  # 挂载本地证书目录至容器
  kafka-gateway:latest

Docker Compose 部署

通过docker-compose部署包含Kafka集群与网关的完整环境（示例）：

yaml
version: '3.8'
services:
  # Kafka集群（示例）
  zookeeper:
    image: docker.xuanyuan.run/confluentinc/cp-zookeeper:latest
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
    ports:
      - "2181:2181"

  kafka1:
    image: docker.xuanyuan.run/confluentinc/cp-kafka:latest
    depends_on:
      - zookeeper
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka1:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

  kafka2:
    image: docker.xuanyuan.run/confluentinc/cp-kafka:latest
    depends_on:
      - zookeeper
    environment:
      KAFKA_BROKER_ID: 2
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka2:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

  # Kafka Gateway服务
  kafka-gateway:
    image: docker.xuanyuan.run/kafka-gateway:latest
    depends_on:
      - kafka1
      - kafka2
    ports:
      - "8080:8080"  # HTTP API端口
      - "9090:9090"  # Prometheus metrics端口
    environment:
      KAFKA_BROKERS: kafka1:9092,kafka2:9092  # Kafka集群地址
      PROTOCOL_TYPE: HTTP  # 启用HTTP协议
      AUTH_ENABLED: true  # 启用Basic Auth
      AUTH_USERNAME: admin
      AUTH_PASSWORD: secure-pass
      LOG_LEVEL: INFO  # 日志级别
      METRICS_PORT: 9090  # metrics暴露端口
    restart: unless-stopped

配置参数说明

基本使用示例

通过HTTP协议发送消息至Kafka

使用Basic Auth认证发送消息到指定Kafka主题：

bash
curl -X POST http://localhost:8080/api/v1/messages \
  -H "Content-Type: application/json" \
  -u admin:secure-pass \
  -d '{
    "topic": "user-tracking",
    "message": "{\"user_id\": 123, \"action\": \"login\"}",
    "partition": 0  # 可选，指定分区
  }'

通过HTTP协议消费Kafka消息

从指定主题消费消息（支持批量拉取）：

bash
curl -X GET "http://localhost:8080/api/v1/messages/user-tracking?limit=10" \
  -u admin:secure-pass

响应示例：

json
{
  "topic": "user-tracking",
  "messages": [
    {"offset": 0, "partition": 0, "timestamp": 1690000000000, "value": "{\"user_id\": 123, \"action\": \"login\"}"},
    {"offset": 1, "partition": 0, "timestamp": 1690000001000, "value": "{\"user_id\": 456, \"action\": \"logout\"}"}
  ]
}

监控指标访问

通过Prometheus metrics端口获取网关运行指标：

bash
curl http://localhost:9090/metrics

关键指标包括：请求吞吐量、消息转发延迟、Kafka连接状态等。