数据一致性与事务处理

数据一致性的基本概念

数据一致性指的是数据库在任何时间点都能保持数据的正确性和完整性。在Mongoose中，数据一致性主要体现在模型定义、数据验证和事务处理三个方面。例如，定义一个用户模型时，可以通过Schema确保字段类型和必填项：

const userSchema = new mongoose.Schema({
  username: {
    type: String,
    required: true,
    unique: true
  },
  age: {
    type: Number,
    min: 18,
    max: 120
  },
  email: {
    type: String,
    match: /^\S+@\S+\.\S+$/
  }
});

Mongoose中的事务处理

MongoDB从4.0版本开始支持多文档事务，Mongoose在此基础上提供了更友好的API。典型的事务使用场景包括银行转账、订单创建等需要多个操作原子执行的业务：

const session = await mongoose.startSession();
session.startTransaction();

try {
  const fromAccount = await Account.findOne({ _id: 'A' }).session(session);
  const toAccount = await Account.findOne({ _id: 'B' }).session(session);
  
  fromAccount.balance -= 100;
  toAccount.balance += 100;
  
  await fromAccount.save();
  await toAccount.save();
  
  await session.commitTransaction();
} catch (error) {
  await session.abortTransaction();
  throw error;
} finally {
  session.endSession();
}

乐观并发控制

Mongoose通过版本号(__v字段)实现乐观锁，防止并发更新导致的数据不一致：

const product = await Product.findById('someId');
product.price = 200;

// 模拟并发修改
const concurrentProduct = await Product.findById('someId');
concurrentProduct.stock -= 1;
await concurrentProduct.save();

try {
  await product.save(); // 抛出VersionError
} catch (err) {
  if (err instanceof mongoose.Error.VersionError) {
    // 处理版本冲突
  }
}

中间件与数据一致性

Mongoose中间件(pre/post钩子)可以在操作前后执行自定义逻辑，确保数据一致性：

orderSchema.pre('save', async function() {
  const product = await Product.findById(this.productId);
  if (product.stock < this.quantity) {
    throw new Error('Insufficient stock');
  }
});

orderSchema.post('save', async function() {
  await Product.updateOne(
    { _id: this.productId },
    { $inc: { stock: -this.quantity } }
  );
});

批量操作的数据一致性

处理批量操作时，使用bulkWrite可以保证更好的性能和数据一致性：

await Character.bulkWrite([
  {
    updateOne: {
      filter: { name: '张三' },
      update: { $set: { age: 30 } }
    }
  },
  {
    updateOne: {
      filter: { name: '李四' },
      update: { $inc: { score: 5 } }
    }
  }
]);

分布式环境下的挑战

在分布式系统中，Mongoose需要配合MongoDB的分片集群处理跨分片事务：

const session = await mongoose.startSession();
session.startTransaction({
  readConcern: { level: 'snapshot' },
  writeConcern: { w: 'majority' }
});

try {
  // 跨分片操作
  await Order.create([{ ... }], { session });
  await Inventory.updateOne({ ... }, { $inc: { ... } }, { session });
  
  await session.commitTransaction();
} catch (error) {
  await session.abortTransaction();
}

错误处理与重试机制

网络波动等临时故障需要实现自动重试逻辑：

const withTransactionRetry = async (txnFunc, session) => {
  let retryCount = 0;
  const MAX_RETRIES = 3;
  
  while (true) {
    try {
      return await txnFunc(session);
    } catch (err) {
      if (err.hasErrorLabel('TransientTransactionError') && 
          retryCount < MAX_RETRIES) {
        retryCount++;
        await new Promise(resolve => setTimeout(resolve, 100 * retryCount));
        continue;
      }
      throw err;
    }
  }
};

读写分离的一致性考量

使用MongoDB的readPreference时需要注意数据同步延迟：

// 写入时使用主节点
await User.create({ name: '王五' }, { writeConcern: { w: 'majority' } });

// 读取时允许从节点
const users = await User.find().read('secondaryPreferred');

数据迁移的原子性

大规模数据迁移需要保证原子性和可回滚：

const migrationSession = await mongoose.startSession();
migrationSession.startTransaction();

try {
  await OldModel.find().session(migrationSession).cursor()
    .eachAsync(async (doc) => {
      await NewModel.create(transform(doc), { session: migrationSession });
      await OldModel.deleteOne({ _id: doc._id }).session(migrationSession);
    }, { parallel: 5 });
    
  await migrationSession.commitTransaction();
} catch (error) {
  await migrationSession.abortTransaction();
}