smartcitiesdata/discovery_streams

Kafka to WebSocket 转发服务镜像

概述

该镜像提供Kafka消息到WebSocket的实时转发能力，通过动态发现Kafka集群中的主题，自动创建对应的WebSocket通道，使客户端可通过WebSocket实时接收Kafka主题消息，简化实时数据流转场景的部署流程。

核心功能

1. 动态Kafka主题发现

• 自动扫描Kafka集群中的主题列表
• 支持配置主题发现间隔时间，适应动态创建/删除的主题场景
• 无需预先配置主题列表，降低人工维护成本

2. WebSocket通道映射

• 每个Kafka主题自动映射为独立的WebSocket通道
• 通道路径规则：ws://<服务地址>:<端口>/<主题名称>
• 支持多客户端同时连接同一通道，实现消息广播

3. 实时消息转发

• 低延迟转发Kafka主题消息至对应WebSocket通道
• 消息格式与Kafka原始消息保持一致（支持JSON、二进制等格式）
• 内置消息缓冲区，避免峰值流量丢失

4. 灵活配置

• 通过环境变量配置Kafka连接参数、WebSocket服务端口等核心参数
• 支持自定义消息序列化/反序列化方式
• 可配置日志级别，便于问题排查

使用场景

1. 实时数据监控系统

适用于需要实时监控Kafka消息的场景，如运维监控、业务指标实时看板等，通过WebSocket将数据推送至前端监控页面。

2. 前端实时数据展示

为Web前端应用提供实时数据来源，如仪表盘、实时排行榜、动态更新的业务数据展示等。

3. 消息分发枢纽

作为中间件连接Kafka消息系统与WebSocket客户端，实现跨系统的实时消息分发，适用于微服务架构中的实时通信场景。

4. 开发调试工具

快速搭建Kafka消息可视化环境，开发人员无需编写复杂消费者代码，通过浏览器即可查看Kafka主题消息。

使用方法

环境变量配置

Docker Run 部署示例

bash
docker run -d \
  --name kafka-ws-bridge \
  -p 8080:8080 \
  -e KAFKA_BOOTSTRAP_SERVERS="kafka-node1:9092,kafka-node2:9092" \
  -e TOPIC_DISCOVERY_INTERVAL=30 \
  -e LOG_LEVEL=DEBUG \
  your-registry/kafka-websocket-bridge:latest

Docker Compose 部署示例

yaml
version: '3.8'
services:
  kafka-ws-bridge:
    image: your-registry/kafka-websocket-bridge:latest
    container_name: kafka-ws-bridge
    ports:
      - "8080:8080"
    environment:
      - KAFKA_BOOTSTRAP_SERVERS=kafka:9092
      - WEBSOCKET_PORT=8080
      - TOPIC_DISCOVERY_INTERVAL=30
      - LOG_LEVEL=INFO
    depends_on:
      - kafka
    restart: unless-stopped

  kafka:
    image: confluentinc/cp-kafka:7.3.0
    container_name: kafka
    ports:
      - "9092:9092"
    environment:
      - KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://kafka:9092
      - KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1
      - KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181
    depends_on:
      - zookeeper

  zookeeper:
    image: confluentinc/cp-zookeeper:7.3.0
    container_name: zookeeper
    environment:
      - ZOOKEEPER_CLIENT_PORT=2181

客户端连接示例

通过WebSocket客户端连接指定主题通道，示例代码（JavaScript）：

javascript
// 连接名为"user-tracking"的Kafka主题通道
const ws = new WebSocket('ws://localhost:8080/user-tracking');

ws.onmessage = function(event) {
  console.log('Received Kafka message:', event.data);
};

ws.onopen = function() {
  console.log('WebSocket connection established');
};

ws.onclose = function() {
  console.log('WebSocket connection closed');
};

注意事项

1. 确保Kafka集群与该服务网络互通，且Kafka引导服务器地址配置正确
2. 生产环境建议配置KAFKA_CONSUMER_GROUP_ID为唯一值，避免与其他消费者组冲突
3. 对于高流量场景，可通过水平扩展服务实例提高并发处理能力
4. 消息格式需与Kafka生产者保持一致，若使用自定义序列化，需确保客户端能正确解析