MongoDB驱动直连金仓：现有Node.js/Python应用“零代码”迁移指南

// 迁移前：连接MongoDB
const { MongoClient } = require('mongodb');
const client = new MongoClient('mongodb://user:pass@mongodb-host:27017/dbname');

// 迁移后：连接金仓（仅改连接串）
const client = new MongoClient('mongodb://user:pass@kingbase-host:27017/dbname');
// ↑ 仅此一处修改，所有业务代码保持不变

应用层：Node.js/Python应用
    ↓
驱动层：MongoDB官方驱动（mongodb/node.js, pymongo）
    ↓
协议层：MongoDB Wire Protocol（TCP 27017端口）
    ↓
金仓协议转换层：documentdb插件
    │
    ├── BSON解析器：将MongoDB BSON转换为金仓JSONB
    ├── 命令路由器：将MongoDB命令映射为SQL操作
    ├── 会话管理器：维护连接状态和事务上下文
    └── 结果转换器：将查询结果转换为BSON格式
    ↓
存储层：金仓JSONB引擎 + 关系存储

# 1. 金仓数据库启用MongoDB兼容模式
# 修改kingbase.conf配置文件
enable_protocol_compat = on
extension_protocol_port = 27017  # MongoDB默认端口
shared_preload_libraries = 'kdb_documentdb_core,kdb_documentdb'

# 2. 创建文档数据库插件
ksql -Usystem -d dbname
CREATE EXTENSION documentdb CASCADE;

# 3. 验证兼容服务
netstat -an | grep 27017  # 应显示监听状态

// app.js - MongoDB连接配置
const mongoose = require('mongoose');

const mongoConfig = {
  uri: 'mongodb://admin:password@mongodb1:27017,mongodb2:27017/appdb?replicaSet=rs0',
  options: {
    useNewUrlParser: true,
    useUnifiedTopology: true,
    readPreference: 'secondaryPreferred',
    maxPoolSize: 50,
    authSource: 'admin',
    authMechanism: 'SCRAM-SHA-256'
  }
};

mongoose.connect(mongoConfig.uri, mongoConfig.options);

// app.js - 金仓连接配置（仅uri变更）
const mongoConfig = {
  // 仅修改主机地址，其他参数完全不变
  uri: 'mongodb://system:123456@kingbase-host:27017/appdb?authSource=admin&authMechanism=SCRAM-SHA-256',
  options: {
    useNewUrlParser: true,
    useUnifiedTopology: true,
    // 注意：金仓读写分离集群支持读偏好设置
    readPreference: 'secondaryPreferred',  // 自动路由到备节点
    maxPoolSize: 50,
    authSource: 'admin',
    authMechanism: 'SCRAM-SHA-256'
  }
};

mongoose.connect(mongoConfig.uri, mongoConfig.options);  // 业务代码零修改

// userService.js - 完全兼容，无需修改
const User = require('./models/User');

class UserService {
  // 创建用户
  async createUser(userData) {
    return await User.create(userData);  // 使用Mongoose ODM
  }
  
  // 复杂查询
  async findActiveUsers(conditions) {
    return await User.find({
      ...conditions,
      status: 'active',
      lastLogin: { $gte: new Date(Date.now() - 30*24*60*60*1000) }
    })
    .sort({ createdAt: -1 })
    .limit(100)
    .skip(0);
  }
  
  // 聚合查询
  async getUserStats() {
    return await User.aggregate([
      { $match: { status: 'active' } },
      { $group: {
        _id: { $dateToString: { format: "%Y-%m", date: "$createdAt" } },
        count: { $sum: 1 },
        avgAge: { $avg: "$age" }
      }},
      { $sort: { _id: 1 } }
    ]);
  }
}

// orderService.js - MongoDB多文档事务
async function createOrder(orderData, items) {
  const session = await mongoose.startSession();
  
  try {
    session.startTransaction();
    
    // 创建订单
    const order = await Order.create([orderData], { session });
    
    // 扣减库存
    for (const item of items) {
      await Product.updateOne(
        { _id: item.productId, stock: { $gte: item.quantity } },
        { $inc: { stock: -item.quantity } },
        { session }
      );
    }
    
    await session.commitTransaction();
    return order;
  } catch (error) {
    await session.abortTransaction();
    throw error;
  } finally {
    session.endSession();
  }
}
// ↑ 金仓完全支持MongoDB事务语法，代码无需修改

// fileService.js - GridFS操作
const { GridFSBucket } = require('mongodb');

async function uploadFile(filename, stream) {
  const bucket = new GridFSBucket(db, {
    bucketName: 'files'
  });
  
  return new Promise((resolve, reject) => {
    const uploadStream = bucket.openUploadStream(filename);
    stream.pipe(uploadStream);
    
    uploadStream.on('finish', resolve);
    uploadStream.on('error', reject);
  });
}
// 金仓原生支持GridFS协议，代码完全兼容

# app.py - 原始MongoDB连接
from pymongo import MongoClient
from pymongo.errors import ConnectionFailure

# MongoDB连接配置
client = MongoClient(
    'mongodb://user:password@mongodb-host:27017/',
    authSource='admin',
    authMechanism='SCRAM-SHA-256',
    maxPoolSize=100,
    readPreference='secondaryPreferred'
)

db = client['mydatabase']
collection = db['mycollection']

# app.py - 金仓连接（仅主机地址变更）
client = MongoClient(
    'mongodb://system:123456@kingbase-host:27017/',  # 仅此处修改
    authSource='admin',
    authMechanism='SCRAM-SHA-256',
    maxPoolSize=100,
    readPreference='secondaryPreferred'  # 金仓支持读写分离
)

db = client['mydatabase']  # 自动创建数据库
collection = db['mycollection']  # 自动创建集合

# models.py - 使用Djongo（Django MongoDB连接器）
from djongo import models

class Blog(models.Model):
    name = models.CharField(max_length=100)
    tags = models.ListField()  # MongoDB特有字段类型
    metadata = models.JSONField()  # 嵌套文档
    
    class Meta:
        db_table = 'blog'
        
# views.py - 业务逻辑完全不变
def get_popular_blogs():
    # 复杂查询
    blogs = Blog.objects.filter(
        tags__contains='python',
        created_at__gte=timezone.now() - timedelta(days=30)
    ).order_by('-view_count')[:10]
    
    # 聚合操作
    from django.db.models import Count
    tag_stats = Blog.objects.all().aggregate([
        {'$unwind': '$tags'},
        {'$group': {'_id': '$tags', 'count': {'$sum': 1}}},
        {'$sort': {'count': -1}},
        {'$limit': 5}
    ])
    
    return blogs, tag_stats

# async_app.py - 异步MongoDB应用
import motor.motor_asyncio
from bson.objectid import ObjectId

# 连接金仓（仅改连接串）
client = motor.motor_asyncio.AsyncIOMotorClient(
    'mongodb://system:123456@kingbase-host:27017/appdb'
)

async def get_user_with_posts(user_id):
    # 关联查询 - 金仓支持$lookup（有限制）
    pipeline = [
        {'$match': {'_id': ObjectId(user_id)}},
        {'$lookup': {
            'from': 'posts',
            'localField': '_id',
            'foreignField': 'author_id',
            'as': 'user_posts'
        }},
        {'$project': {
            'name': 1,
            'email': 1,
            'post_count': {'$size': '$user_posts'},
            'recent_posts': {'$slice': ['$user_posts', 5]}
        }}
    ]
    
    cursor = client.appdb.users.aggregate(pipeline)
    return await cursor.to_list(length=1)

# 1. 使用mongodump从MongoDB导出
mongodump --uri="mongodb://user:pass@mongodb-host:27017/sourcedb" \
  --out=/backup/mongodb_dump \
  --gzip

# 2. 使用mongorestore导入金仓
mongorestore --uri="mongodb://system:123456@kingbase-host:27017/targetdb" \
  /backup/mongodb_dump \
  --gzip \
  --drop  # 可选：清空目标库

# 3. 使用KFS实现增量同步（零停机）
# 建立MongoDB到金仓的实时同步
kfs_start_sync \
  --source-type=mongodb \
  --source-uri="mongodb://mongodb-host:27017" \
  --target-type=kingbase \
  --target-uri="kingbase://system:123456@kingbase-host:54321" \
  --tables="sourcedb.*"

# validation_script.py - 自动数据比对
import pymongo
from pymongo import MongoClient
import hashlib

def compare_collections(mongo_uri, kingbase_uri, db_name, collection_name):
    # 连接两个数据库
    mongo_client = MongoClient(mongo_uri)
    kingbase_client = MongoClient(kingbase_uri)  # 相同驱动！
    
    mongo_coll = mongo_client[db_name][collection_name]
    kingbase_coll = kingbase_client[db_name][collection_name]
    
    # 比对记录数
    mongo_count = mongo_coll.count_documents({})
    kingbase_count = kingbase_coll.count_documents({})
    
    print(f"记录数比对: MongoDB={mongo_count}, 金仓={kingbase_count}")
    
    # 抽样比对数据内容
    sample_size = min(1000, mongo_count)
    mongo_sample = list(mongo_coll.aggregate([
        {'$sample': {'size': sample_size}},
        {'$sort': {'_id': 1}}
    ]))
    
    kingbase_sample = list(kingbase_coll.aggregate([
        {'$sample': {'size': sample_size}},
        {'$sort': {'_id': 1}}
    ]))
    
    # 计算哈希比对
    for i in range(sample_size):
        mongo_hash = hashlib.md5(str(mongo_sample[i]).encode()).hexdigest()
        kingbase_hash = hashlib.md5(str(kingbase_sample[i]).encode()).hexdigest()
        
        if mongo_hash != kingbase_hash:
            print(f"数据不一致: _id={mongo_sample[i]['_id']}")
            return False
    
    print("数据一致性验证通过！")
    return True

// 金仓JSONB索引建议
// 1. 高频查询字段创建GIN索引
db.collection.createIndex({ "user.email": 1 });  // 自动转换为JSONB索引

// 2. 复杂查询路径创建jsonb_path_ops索引
// 金仓自动优化：将MongoDB索引转换为最优JSONB索引类型

// 3. 查看索引使用情况
const explainResult = db.collection.find(
  { "status": "active", "metadata.score": { $gt: 80 } }
).explain("executionStats");

console.log(explainResult.executionStats);

// Node.js连接池最佳配置
const client = new MongoClient('mongodb://kingbase-host:27017', {
  poolSize: 50,  // 根据业务并发调整
  minPoolSize: 10,
  maxPoolSize: 100,
  maxIdleTimeMS: 60000,
  waitQueueTimeoutMS: 5000,
  connectTimeoutMS: 10000,
  socketTimeoutMS: 45000,
  
  // 金仓特有优化参数
  readPreference: 'secondaryPreferred',  // 利用读写分离
  readConcern: { level: 'local' },  // 读一致性级别
  writeConcern: { w: 'majority', j: true }  // 写确认级别
});

# 金仓特有监控命令
# 1. 查看MongoDB协议连接
SELECT * FROM sys_protocol_connections WHERE protocol_type = 'mongodb';

# 2. 监控查询性能
SELECT * FROM sys_mongodb_slow_queries 
WHERE duration > 1000  -- 超过1秒的查询
ORDER BY duration DESC;

# 3. 索引使用统计
SELECT * FROM sys_jsonb_index_usage 
WHERE collection_name = 'users';

// 原代码使用$facet（金仓暂不支持）
const result = await collection.aggregate([
  { $facet: {
    'categoryStats': [...],
    'priceStats': [...]
  }}
]);

// 解决方案：拆分为多个查询
const [categoryStats, priceStats] = await Promise.all([
  collection.aggregate([...]),  // 分类统计
  collection.aggregate([...])   // 价格统计
]);

# 原MongoDB地理查询
results = collection.find({
  "location": {
    "$near": {
      "$geometry": {
        "type": "Point",
        "coordinates": [longitude, latitude]
      },
      "$maxDistance": 5000
    }
  }
})

# 金仓解决方案：使用PostGIS扩展
# 1. 启用PostGIS扩展
# 2. 将地理数据转换为金仓GIS格式
# 3. 使用ST_DWithin等函数查询