### 使用 Docker Compose 搭建 MongoDB 副本集 + 变更流环境 以下是一个完整的解决方案，使用 Docker Compose 搭建一个包含三个节点的 MongoDB 副本集，并启用变更流功能。 #### 1. 创建 `docker-compose.yml` 文件 ```yaml version: '3.8' services: mongo1: image: mongo:6.0 container_name: mongo1 command: mongod --replSet rs0 --bind_ip_all --port 27017 ports: - "27017:27017" volumes: - ./data/mongo1:/data/db environment: MONGO_INITDB_ROOT_USERNAME: root MONGO_INITDB_ROOT_PASSWORD: example mongo2: image: mongo:6.0 container_name: mongo2 command: mongod --replSet rs0 --bind_ip_all --port 27017 expose: - "27017" volumes: - ./data/mongo2:/data/db environment: MONGO_INITDB_ROOT_USERNAME: root MONGO_INITDB_ROOT_PASSWORD: example mongo3: image: mongo:6.0 container_name: mongo3 command: mongod --replSet rs0 --bind_ip_all --port 27017 expose: - "27017" volumes: - ./data/mongo3:/data/db environment: MONGO_INITDB_ROOT_USERNAME: root MONGO_INITDB_ROOT_PASSWORD: example mongosetup: image: mongo:6.0 container_name: mongosetup depends_on: - mongo1 - mongo2 - mongo3 entrypoint: ["sh", "-c", "sleep 10 && mongo --host mongo1:27017 -u root -p example --eval \"rs.initiate({ _id: 'rs0', members: [ { _id: 0, host: 'mongo1:27017' }, { _id: 1, host: 'mongo2:27017' }, { _id: 2, host: 'mongo3:27017' } ] });\""] networks: default: driver: bridge ``` --- ### 代码解释 1. **服务定义**： - `mongo1`, `mongo2`, `mongo3`：分别代表副本集中的三个节点。 - 每个节点通过 `command` 参数指定为副本集成员，并绑定所有 IP 地址 (`--bind_ip_all`)。 - 数据持久化通过 `volumes` 挂载到主机的 `./data/mongoX` 目录。 2. **初始化副本集**： - `mongosetup` 服务负责在容器启动后初始化副本集。 - 使用 `mongo` 命令行工具执行 `rs.initiate()` 方法，配置副本集成员。 3. **网络配置**： - 默认使用 `bridge` 网络，确保所有容器可以互相通信。 4. **端口映射**： - `mongo1` 的 27017 端口被映射到主机的 27017 端口，便于外部访问。 - 其他节点仅暴露给内部网络。 --- ### 启动环境 1. **创建目录结构**： ```bash mkdir -p ./data/mongo1 ./data/mongo2 ./data/mongo3 ``` 2. **启动容器**： ```bash docker-compose up -d ``` 3. **验证副本集状态**： 进入 `mongo1` 容器并检查副本集状态： ```bash docker exec -it mongo1 bash mongo -u root -p example ``` 在 MongoDB shell 中运行： ```javascript rs.status(); ``` --- ### 配置变更流监听器 以下是一个 Python 脚本，用于监听 MongoDB 的变更流： ```python from pymongo import MongoClient from pymongo.errors import ConnectionFailure, ServerSelectionTimeoutError import logging import time # 配置日志系统 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s' ) logger = logging.getLogger('MongoDB_Change_Stream') def connect_to_mongodb(): """连接到 MongoDB 副本集""" try: client = MongoClient( "mongodb://root:example@mongo1:27017,mongo2:27017,mongo3:27017/?replicaSet=rs0", serverSelectionTimeoutMS=5000 ) client.admin.command('ping') # 测试连接 logger.info("成功连接到 MongoDB 副本集") return client except (ConnectionFailure, ServerSelectionTimeoutError) as e: logger.error(f"无法连接到 MongoDB: {str(e)}") return None def watch_change_stream(client, db_name="test_db", collection_name="test_collection"): """监听变更流""" db = client[db_name] collection = db[collection_name] pipeline = [{"$match": {"operationType": {"$in": ["insert", "update", "delete"]}}}] try: with collection.watch(pipeline) as stream: logger.info("开始监听变更流...") for change in stream: logger.info(f"检测到变更: {change}") except Exception as e: logger.error(f"变更流监听失败: {str(e)}") if __name__ == "__main__": client = connect_to_mongodb() if client: try: watch_change_stream(client) except KeyboardInterrupt: logger.info("用户中断监听") finally: client.close() logger.info("关闭 MongoDB 连接") ``` --- ### 代码解释 1. **连接到 MongoDB**： - 使用 `MongoClient` 连接到副本集，指定用户名、密码和副本集名称。 - 测试连接是否成功（`admin.command('ping')`）。 2. **监听变更流**： - 使用 `collection.watch()` 方法创建变更流监听器。 - 配置管道过滤插入、更新和删除操作。 3. **日志记录**： - 记录每次变更事件的详细信息。 --- ### 测试变更流 1. **插入数据**： 在另一个终端中，连接到 MongoDB 并插入测试数据： ```bash docker exec -it mongo1 bash mongo -u root -p example ``` 在 MongoDB shell 中运行： ```javascript use test_db; db.test_collection.insertOne({ name: "Alice", age: 25 }); ``` 2. **观察日志**： 查看 Python 脚本的日志输出，确认变更流已捕获插入事件。 --- ### 注意事项 1. **副本集健康检查**： 确保所有节点正常运行，并且副本集状态良好（`rs.status()`）。 2. **变更流权限**： 确保 MongoDB 用户具有 `find` 和 `changeStream` 权限。 3. **网络延迟**： 如果节点分布在不同机器上，可能需要调整网络设置以减少延迟。 ---