本站支持搜索的镜像仓库：Docker Hub、gcr.io、ghcr.io、quay.io、k8s.gcr.io、registry.gcr.io、elastic.co、mcr.microsoft.com

etf1/kafka-mongo-watcher

监听MongoDB集合的oplog事件（插入、更新、删除等）并将其推送到指定Kafka主题，同时支持重播模式以首次初始化集合所有文档到Kafka的工具

0 次下载activeetf1镜像

🚀轩辕镜像专业版更稳定💎一键安装 Docker 配置镜像源

中文简介版本下载

🚀轩辕镜像专业版更稳定💎一键安装 Docker 配置镜像源

Kafka MongoDB Watcher Docker镜像文档

镜像概述与主要用途

Kafka MongoDB Watcher是一个用于监听MongoDB集合操作日志（oplog）事件（如插入、更新、删除等）并将其分发到指定Kafka主题的工具。同时支持重播模式，可在首次使用时将集合中的所有文档初始化到Kafka主题中。该Docker镜像封装了工具的运行环境，简化部署流程。

核心功能与特性

oplog事件监听：实时捕获MongoDB集合的插入、更新、删除等操作事件
重播模式：支持一次性将集合所有文档发送到Kafka主题，适用于首次初始化
灵活配置：通过环境变量自定义MongoDB连接、Kafka连接、过滤规则等参数
Prometheus指标：暴露监控指标，便于性能和状态监控
多模式运行：支持同时启用监听和重播模式，满足不同数据同步需求

使用场景与适用范围

数据同步：将MongoDB数据变更实时同步到Kafka，用于下游系统消费
实时数据流处理：作为数据管道组件，为流处理系统（如Spark Streaming、Flink）提供MongoDB变更数据
事件驱动架构：基于MongoDB数据变更触发Kafka消息，实现微服务间的事件通信
数据备份与回放：通过重播模式创建数据快照，用于数据备份或历史数据回放

详细使用方法

Docker镜像获取

镜像已发布至Docker Hub，可直接拉取：

docker pull etf1/kafka-mongo-watcher:latest

基本运行示例

使用docker run命令启动，通过环境变量配置参数：

docker run \
  -e "KAFKA_MONGO_WATCHER_REPLAY=true" \
  -e "KAFKA_MONGO_WATCHER_MONGODB_URI=mongodb://root:password@mongo:27017/mydb?replicaSet=rs0" \
  -e "KAFKA_MONGO_WATCHER_MONGODB_DATABASE_NAME=mydb" \
  -e "KAFKA_MONGO_WATCHER_MONGODB_COLLECTION_NAME=mycollection" \
  -e "KAFKA_MONGO_WATCHER_KAFKA_BOOTSTRAP_SERVERS=kafka:9092" \
  -e "KAFKA_MONGO_WATCHER_KAFKA_TOPIC=mongodb-events" \
  etf1/kafka-mongo-watcher:latest

docker-compose配置示例

version: '3'
services:
  kafka-mongo-watcher:
    image: etf1/kafka-mongo-watcher:latest
    environment:
      - KAFKA_MONGO_WATCHER_REPLAY=true
      - KAFKA_MONGO_WATCHER_MONGODB_URI=mongodb://root:password@mongo:27017/mydb?replicaSet=rs0
      - KAFKA_MONGO_WATCHER_MONGODB_DATABASE_NAME=mydb
      - KAFKA_MONGO_WATCHER_MONGODB_COLLECTION_NAME=mycollection
      - KAFKA_MONGO_WATCHER_KAFKA_BOOTSTRAP_SERVERS=kafka:9092
      - KAFKA_MONGO_WATCHER_KAFKA_TOPIC=mongodb-events
      - KAFKA_MONGO_WATCHER_LOG_LEVEL=info
    depends_on:
      - mongo
      - kafka

  mongo:
    image: mongo:5
    command: --replSet rs0
    environment:
      - MONGO_INITDB_ROOT_USERNAME=root
      - MONGO_INITDB_ROOT_PASSWORD=password

  kafka:
    image: confluentinc/cp-kafka:7.0.0
    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.0.0
    environment:
      - ZOOKEEPER_CLIENT_PORT=2181

配置参数说明

可通过环境变量配置镜像，支持多源配置（.env文件、环境变量、命令行参数，后者覆盖前者）。以下是主要配置变量：

环境变量	类型	描述	示例值	默认值
KAFKA_MONGO_WATCHER_CUSTOM_PIPELINE	string	自定义过滤管道，支持重播和监听模式	`[ { "$match": { "fullDocument.is_active": true } }, { $addFields: { "custom-field": "custom-value" } } ]`	-
KAFKA_MONGO_WATCHER_REPLAY	bool	是否一次性发送集合所有文档	`true`	`false`
KAFKA_MONGO_WATCHER_MONGODB_URI	string	MongoDB连接URI	`mongodb://root:password@mongo:27017/mydb?replicaSet=rs0`	`mongodb://root:toor@127.0.0.1:27011,...`
KAFKA_MONGO_WATCHER_MONGODB_COLLECTION_NAME	string	要监听的MongoDB集合名	`users`	`items`
KAFKA_MONGO_WATCHER_MONGODB_DATABASE_NAME	string	MongoDB数据库名	`mydb`	`watcher`
KAFKA_MONGO_WATCHER_MONGODB_SERVER_SELECTION_TIMEOUT	duration	MongoDB服务器选择超时时间	`5s`	`2s`
KAFKA_MONGO_WATCHER_MONGODB_OPTION_BATCH_SIZE	integer	MongoDB监听批处理大小（0为不批处理）	`100`	`0`
KAFKA_MONGO_WATCHER_MONGODB_OPTION_FULL_DOCUMENT	boolean	是否在oplog中返回完整文档	`false`	`true`
KAFKA_MONGO_WATCHER_MONGODB_OPTION_MAX_AWAIT_TIME	duration	等待新oplog的最大时间（0为不停止）	`30s`	`0`
KAFKA_MONGO_WATCHER_MONGODB_OPTION_RESUME_AFTER	string	变更流的起始逻辑点	`{"_data": "82650..."}`	-
KAFKA_MONGO_WATCHER_MONGODB_OPTION_START_AT_OPERATION_TIME_I	uint32	变更流起始时间戳（递增值）	`12345`	-
KAFKA_MONGO_WATCHER_MONGODB_OPTION_START_AT_OPERATION_TIME_T	uint32	变更流起始时间戳（时间值）	`1620000000`	-
KAFKA_MONGO_WATCHER_MONGODB_OPTION_WATCH_MAX_RETRIES	integer	监听集合的最大重试次数（0为禁用重试）	`5`	`3`
KAFKA_MONGO_WATCHER_MONGODB_OPTION_WATCH_RETRY_DELAY	duration	监听重试间隔	`1s`	`500ms`
KAFKA_MONGO_WATCHER_KAFKA_BOOTSTRAP_SERVERS	string	Kafka引导服务器列表	`kafka1:9092,kafka2:9092`	`127.0.0.1:9092`
KAFKA_MONGO_WATCHER_KAFKA_TOPIC	string	目标Kafka主题名	`mongodb-events`	`kafka-mongo-watcher`
KAFKA_MONGO_WATCHER_KAFKA_PRODUCE_CHANNEL_SIZE	integer	Kafka生产者内部通道最大容量	`20000`	`10000`
KAFKA_MONGO_WATCHER_LOG_CLI_VERBOSE	boolean	是否启用详细日志	`false`	`true`
KAFKA_MONGO_WATCHER_LOG_LEVEL	string	日志级别	`debug`	`info`
KAFKA_MONGO_WATCHER_GRAYLOG_ENDPOINT	string	Graylog日志推送端点	`udp://graylog:12201`	-
KAFKA_MONGO_WATCHER_HTTP_TECH_ADDR	string	技术HTTP服务器监听地址	`:8080`	`:8001`
KAFKA_MONGO_WATCHER_PRINT_CONFIG	boolean	启动时是否打印配置	`false`	`true`
KAFKA_MONGO_WATCHER_PPROF_ENABLED	boolean	是否启用Go pprof调试	`false`	`true`