mongo分片搭建

环境如下：

138 gbaseNode1

139 gbaseNode2

140

分片集群如下：

创建复制集

创建目录

在三台机器上创建目录：

mkdir -p /data/mongo/share/data/db
mkdir -p /data/mongo/share/data/db2
mkdir -p /data/mongo/share/conf
mkdir -p /data/mongo/share/pids
mkdir -p /data/mongo/share/logs

创建配置文件

在三台机器的/data/mongo/share/conf 目录下，添加分片0复制集的配置文件：mongo.conf ：

#数据保存路径
dbpath=/data/mongo/share/data/db
#日志保存路径
logpath=/data/mongo/share/logs/mongo.log
#进程描述文件
pidfilepath=/data/mongo/share/pids/mongo.pid
#日志追加写入
logappend=true
#复制集名称
replSet=rs0
bind_ip_all=true
#mongo端口
port=27017
#操作日志容量
oplogSize=10000
#开启子进程
fork=true
#代表当前节点是一个shard节点
shardsvr=true

在三台机器的/data/mongo/share/conf 目录下，添加分片1复制集的配置文件：mongo2.conf ：

#数据保存路径
dbpath=/data/mongo/share/data/db2
#日志保存路径
logpath=/data/mongo/share/logs/mongo2.log
#进程描述文件
pidfilepath=/data/mongo/share/pids/mongo2.pid
#日志追加写入
logappend=true
#复制集名称
replSet=rs1
bind_ip_all=true
#mongo端口
port=27018
#操作日志容量
oplogSize=10000
#开启子进程
fork=true
#代表当前节点是一个shard节点
shardsvr=true

启动mongod服务

在三台机器上启动两个分片集群的mongod服务：

#分片0(rs0)
/data/mongodb-linux-x86_64-rhel80-5.0.31/bin/mongod -f /data/mongo/share/conf/mongo.conf
#分片1(rs1)
/data/mongodb-linux-x86_64-rhel80-5.0.31/bin/mongod -f /data/mongo/share/conf/mongo2.conf

初始化分片集

从138分别登录和初始化复制集

# 登录rs0
/data/mongodb-linux-x86_64-rhel80-5.0.31/bin/mongo --host 10.25.245.138 --port 27017
# 初始化
rs.initiate({
    _id:"rs0",
    members:[
        {_id:0, host:"10.25.245.138:27017", priority:3},
        {_id:1, host:"10.25.245.139:27017", priority:2},
        {_id:2, host:"10.25.245.140:27017", arbiterOnly:true}
    ]
});


# 登录rs1
/data/mongodb-linux-x86_64-rhel80-5.0.31/bin/mongo --host 10.25.245.138 --port 27018
# 初始化
rs.initiate({
    _id:"rs1",
    members:[
        {_id:0, host:"10.25.245.138:27018", priority:3},
        {_id:1, host:"10.25.245.139:27018", priority:2},
        {_id:2, host:"10.25.245.140:27018", arbiterOnly:true}
    ]
});

搭建Config Server

创建目录

在三台机器上创建config目录：

mkdir -p /data/mongo/config-server/data/db
mkdir -p /data/mongo/config-server/conf
mkdir -p /data/mongo/config-server/pids
mkdir -p /data/mongo/config-server/logs

创建配置文件

在三台机器上创建配置文件：

cat /data/mongo/config-server/conf/mongo-cfg.conf

dbpath=/data/mongo/config-server/data/db
logpath=/data/mongo/config-server/logs/mongo.log
pidfilepath=/data/mongo/config-server/pids/mongo.pid
logappend=true
replSet=rs_conf
bind_ip_all=true
port=28018
oplogSize=10000
fork=true
#代表当前节点是一个config节点
configsvr=true

启动与初始化复制集

在三台机器上启动复制集

/data/mongodb-linux-x86_64-rhel80-5.0.31/bin/mongod -f /data/mongo/config-server/conf/mongo-cfg.conf

从138上登录并初始化复制集

# 登录
/data/mongodb-linux-x86_64-rhel80-5.0.31/bin/mongo --host 10.25.245.138 --port 28018

# 初始化
rs.initiate({
    _id:"rs_conf",
    configsvr: true,
    members:[
        {_id:0, host:"10.25.245.138:28018", priority:2},
        {_id:1, host:"10.25.245.139:28018", priority:1},
        {_id:2, host:"10.25.245.140:28018", priority:1}
    ]
});

搭建router

本案例只配置了一个Router，生产环境通常配置多个。

创建目录

在138上配置:

mkdir -p /data/mongo/router/conf
mkdir -p /data/mongo/router/pids
mkdir -p /data/mongo/router/logs

创建配置文件

cat /data/mongo/router/conf/mongo-router.conf

configdb=rs_conf/10.25.245.138:28018,10.25.245.139:28018,10.25.245.140:28018
logpath=/data/mongo/router/logs/mongos.log
pidfilepath=/data/mongo/router/pids/mongos.pid
port=29019
fork=true
bind_ip_all=true

启动router服务

/data/mongodb-linux-x86_64-rhel80-5.0.31/bin/mongos -f /data/mongo/router/conf/mongo-router.conf

添加分片信息

# 连接router
/data/mongodb-linux-x86_64-rhel80-5.0.31/bin/mongo --host 10.25.245.138 --port 29019

# 切换到admin库
use admin

# 添加分片信息
#分片0
db.runCommand({addShard: "rs0/10.25.245.138:27017,10.25.245.139:27017,10.25.245.140:27017", name: "share0"});
#分片1
db.runCommand({addShard: "rs1/10.25.245.138:27018,10.25.245.139:27018,10.25.245.140:27018", name: "share1"});

# 查看分片
db.runCommand({listshards: 1})

# 查看分片状态
sh.status()

测试

# 连接到router节点
/data/mongodb-linux-x86_64-rhel80-5.0.31/bin/mongo --host 10.25.245.138 --port 29019

# 切换到 admin 库。对测试库 xwjdb 开启分片功能
use admin
sh.enableSharding("xwjdb")

# 对集合 account 分片，分片键为 nickname，分片策略为哈希：
sh.shardCollection("xwjdb.account", {"nickname": "hashed"})

# 切换到 xwjdb 库。向 account 集合中，循环插入1000条数据：
use xwjdb
for(var i=1; i<=1000; i++){db.account.insert({_id: i+"", nickname: "xwj"+i})}

# 分别连接 分片0 和 分片1 节点，并切换到 xwjdb 库。查询 account 集合数据条数：
use xwjdb
db.account.count()

插入测试数据

测试脚本：

# -*- coding: utf-8 -*-
"""
向 MongoDB 写入 测试数据的脚本
运行环境：Python 3.6 + pymongo + dataclasses
MongoDB 地址：192.168.75.66:27017
"""

from datetime import datetime, timedelta
import random
import string
from dataclasses import dataclass, asdict
import pymongo
import uuid

# =============================
# 配置参数
# =============================

MONGO_HOST = "192.168.75.66"
MONGO_PORT = 27017
DATABASE_NAME = "testdb"           # 数据库名（会自动创建）
COLLECTION_NAME = "testdata"       # 集合名

#TARGET_TOTAL_DOCS = 2_000_000      # 约 1000 万条，每条 ~1KB => 总共 ~2GB
TARGET_TOTAL_DOCS = 500_000      # 约 1000 万条，每条 ~1KB => 总共 ~500M
BATCH_SIZE = 1000                  # 每批插入 1000 条

# 字段池
FIRST_NAMES = ["Alice", "Bob", "Charlie", "Diana", "Eve", "Frank", "Grace", "Henry", "Ivy", "Jack"]
LAST_NAMES = ["Smith", "Johnson", "Williams", "Brown", "Jones", "Garcia", "Miller", "Davis", "Rodriguez", "Martinez"]
DOMAINS = ["example.com", "mail.com", "user.net", "test.org", "demo.io"]
DEPARTMENTS = ["Engineering", "Sales", "HR", "Marketing", "Finance", "Support", "Operations"]
STATUSES = ["active", "inactive", "pending", "blocked"]

# 用于生成随机文本
def randstr(n):
    return ''.join(random.choices(string.ascii_letters + string.digits, k=n))

def randdate(start_year=2010):
    year = random.randint(start_year, 2025)
    month = random.randint(1, 12)
    day = random.randint(1, 28)
    return datetime(year, month, day)

def randemail():
    user = randstr(random.randint(6, 12)).lower()
    domain = random.choice(DOMAINS)
    return f"{user}@{domain}"

# =============================
# 使用 dataclass 定义数据结构
# =============================

@dataclass
class TestRecord:
    _id: str
    first_name: str
    last_name: str
    email: str
    department: str
    salary: float
    status: str
    login_count: int
    last_login: datetime
    created_at: datetime
    metadata: dict  # 加大体积用

# =============================
# 主函数
# =============================

def main():
    print("开始连接 MongoDB...")
    client = pymongo.MongoClient(
        host=MONGO_HOST,
        port=MONGO_PORT,
        serverSelectionTimeoutMS=10000,
        socketTimeoutMS=30000,
        connectTimeoutMS=10000
    )

    try:
        client.admin.command('ping')
        print("成功连接到 MongoDB")
    except Exception as e:
        print(f"连接失败: {e}")
        return

    db = client[DATABASE_NAME]
    coll = db[COLLECTION_NAME]

    print(f"开始向 {DATABASE_NAME}.{COLLECTION_NAME} 插入 {TARGET_TOTAL_DOCS:,} 条测试数据...")

    start_time = datetime.now()
    inserted_count = 0

    for batch_start in range(0, TARGET_TOTAL_DOCS, BATCH_SIZE):
        batch_end = min(batch_start + BATCH_SIZE, TARGET_TOTAL_DOCS)
        batch = []

        for i in range(batch_start + 1, batch_end + 1):
            # 构造唯一 _id
            record_id = str(uuid.uuid4())

            # 创建一条测试记录
            record = TestRecord(
                _id=record_id,
                first_name=random.choice(FIRST_NAMES),
                last_name=random.choice(LAST_NAMES),
                email=randemail(),
                department=random.choice(DEPARTMENTS),
                salary=round(random.uniform(3000, 150000), 2),
                status=random.choice(STATUSES),
                login_count=random.randint(0, 500),
                last_login=randdate(2020),
                created_at=randdate(2010),
                metadata={
                    'tags': [randstr(5) for _ in range(3)],
                    'note': randstr(200),  # 占用空间的关键字段（拉高至 ~1KB）
                    'ip': f"192.168.{random.randint(1,255)}.{random.randint(1,255)}",
                    'agent': f"Mozilla/{randstr(8)} Chrome/{random.randint(80,120)}.0"
                }
            )
            batch.append(asdict(record))  # 转为字典以便插入

        # 批量写入
        try:
            result = coll.insert_many(batch, ordered=False)  # 出错不停止
            inserted_count += len(result.inserted_ids)
        except pymongo.errors.BulkWriteError as bwe:
            # 忽略重复键等问题
            failed = len(bwe.details['writeErrors'])
            print(f"批次写入部分失败，{failed} 条出错 | 继续...")
        except Exception as e:
            print(f"写入异常: {e} | 继续下一批...")

        # 输出进度（每 10 万条）
        if (batch_end % 100_000 == 0) or (batch_end == TARGET_TOTAL_DOCS):
            elapsed = (datetime.now() - start_time).total_seconds()
            rate = inserted_count / elapsed if elapsed > 0 else 0
            print(f"进度: {batch_end:,}/{TARGET_TOTAL_DOCS:,} | "
                  f"速率: {rate:.0f} docs/s | "
                  f"已用时: {int(elapsed)}s")

    total_time = (datetime.now() - start_time).total_seconds()
    print(f"\n完成！总计插入 {inserted_count:,} 条文档，耗时 {total_time:.2f} 秒。")
    print(f"平均写入速度: {inserted_count / total_time:.0f} 文档/秒")
    print(f"数据库: {DATABASE_NAME}, 集合: {COLLECTION_NAME}")
    print(f"提示：可通过 mongostat 或数据库监控查看写入情况。")

    client.close()


if __name__ == "__main__":
    main()

配置hash分片：

# 连接到router节点
/data/mongodb-linux-x86_64-rhel80-5.0.31/bin/mongo --host 10.25.245.138 --port 29019

# 切换到 admin 库。对测试库 testdb 开启分片功能
use admin
mongos> sh.enableSharding("testdb")
sh.shardCollection("testdb.testdata", {"first_name": "hashed"})

# 执行脚本
python3 insert_test_data_mongo.py