CQRS All In One

Command Query Responsibility Segregation / 命令查询责任分离

CQRS 核心思想是将应用程序分为两部分：命令（写操作/更新） 和 查询（读操作），并为这两类操作使用不同的数据模型和存储结构，以提升系统的性能、安全性和扩展性。

• 命令（Command）：负责引起系统状态变化的操作（如新增、更新、删除）。它侧重于业务逻辑和数据一致性，通常不返回数据。
• 查询（Query）：负责获取系统状态的操作（如数据查找、报表生成）。它只读，不改变数据状态，侧重于高效率和高并发。

1. 单个数据存储中的单独模型

2. 不同数据存储中的单独模型

为什么要使用 CQRS？

传统的 CRUD（增删改查）系统往往使用同一个数据模型进行读写，这在复杂业务或高并发场景下容易遇到瓶颈。
CQRS 带来了以下优势：
独立扩展：读和写可以部署在不同的服务器集群上，独立进行横向扩展（例如：读取请求多时，可增加只读数据库副本来分摊流量）。
模型简化：写入端只需关注复杂的业务规则和数据完整性；读取端则可针对具体页面或报表优化，避免复杂的表连接（JOIN）或 ORM 映射。
安全性提升：可以将写逻辑与读逻辑严格隔离，仅允许特定角色访问命令服务。

应用场景

复杂领域驱动设计（DDD）：在大型系统中，写入端严格遵循 DDD 的聚合根和实体规则，而查询端则绕过复杂模型，直接执行 SQL 查询以提高性能。
读写压力悬殊的系统：电商平台、社交媒体等读取请求远大于写入请求的应用。
事件溯源（Event Sourcing）：在记录业务事件的系统中，通常结合 CQRS 使用，通过事件流异步更新读取端数据库。

注意的挑战

数据一致性：因为读写分离，系统通常采用“最终一致性”。写入端的数据可能需要一定的时间（通过消息队列或异步事件）同步到查询端，对实时性要求极高的场景需妥善处理。
架构复杂度增加：系统需要维护两套不同的模型甚至数据库，增加了初期开发和维护成本。

demos

Express.js 手动实现轻量级 CQRS 架构

src/
├── write-side/         # 写端：包含复杂的业务逻辑
│   ├── controllers/    # 接收 POST/PUT/DELETE 请求
│   └── models/         # 关系型数据库模型（如 MySQL），关注核心业务规则
├── read-side/          # 读端：关注高性能查询
│   ├── controllers/    # 接收 GET 请求
│   └── models/         # 缓存或无关系型数据库（如 Redis/MongoDB），存储扁平化视图
└── app.js

Write Side

// src/write-side/controllers/order.controller.js
const db = require('../../shared/write-db-connection'); // 主数据库连接

exports.createOrder = async (req, res) => {
  try {
    const { userId, items } = req.body;
    // 1. 执行复杂的写库事务（扣减库存、计算折扣等）
    const orderId = await db.transaction(async (trx) => {
      const [id] = await trx('orders').insert({ user_id: userId, status: 'PENDING' });
      // 复杂的扣减库存逻辑
      // ...
      return id;
    });
    // 2. 异步同步到读库（模拟事件发布）
    process.nextTick(async () => {
      const readDb = require('../../shared/read-db-connection');
      // 直接插入一条适合前端展示的扁平化 JSON 数据到读库（如 MongoDB）
      await readDb.collection('order_views').insertOne({
        orderId, 
        userId, 
        items, 
        status: 'PENDING', 
        createdAt: new Date()
      });
    });
    res.status(201).json({ success: true, orderId });
  } catch (error) {
    res.status(500).json({ error: error.message });
  }
};

Read Side

// src/read-side/controllers/order.controller.js
const readDb = require('../../shared/read-db-connection'); // 只读数据库或缓存连接

exports.getOrderDetails = async (req, res) => {
  try {
    const { id } = req.params;
    // 高性能查询：不进行任何多表 JOIN，直接从面向视图的数据库中单表读取
    const orderView = await readDb.collection('order_views').findOne({ orderId: id });
    if (!orderView) return res.status(404).json({ message: '订单不存在' });
    res.json(orderView);
  } catch (error) {
    res.status(500).json({ error: error.message });
  }
};

路由聚合

// src/app.js
const express = require('express');
const app = express();
app.use(express.json());

const writeController = require('./write-side/controllers/order.controller');
const readController = require('./read-side/controllers/order.controller');

// 严格遵循：POST/PUT/DELETE 路由走写端
app.post('/api/orders', writeController.createOrder);

// GET 路由走读端
app.get('/api/orders/:id', readController.getOrderDetails);

app.listen(3000, () => console.log('CQRS Express Server 运行在端口 3000'));

NestJS（原生支持 CQRS）

Entity

// user.entity.ts
export class User {
  constructor(public id: string, public name: string, public email: string) {}
}

Command

// create-user.command.ts (定义命令数据结构)
export class CreateUserCommand {
  constructor(public readonly name: string, public readonly email: string) {}
}

// create-user.handler.ts (处理命令的业务逻辑)
import { CommandHandler, ICommandHandler } from '@nestjs/cqrs';
import { CreateUserCommand } from './create-user.command';

@CommandHandler(CreateUserCommand)
export class CreateUserHandler implements ICommandHandler<CreateUserCommand> {
  async execute(command: CreateUserCommand): Promise<string> {
    const { name, email } = command;
    const userId = Math.random().toString(36).substr(2, 9);
    // 写入主数据库（写库，如 PostgreSQL/MySQL）
    console.log(`正在写入写库: 插入用户 ${name}`); 
    // 实际项目中，这里通常会触发一个内部事件，用于同步数据到读库
    return userId; 
  }
}

Query

// get-user.query.ts (定义查询条件)
export class GetUserByIdQuery {
  constructor(public readonly userId: string) {}
}

// get-user.handler.ts (处理查询的高效逻辑)
import { IQueryHandler, QueryHandler } from '@nestjs/cqrs';
import { GetUserByIdQuery } from './get-user.query';

@QueryHandler(GetUserByIdQuery)
export class GetUserByIdHandler implements IQueryHandler<GetUserByIdQuery> {
  async execute(query: GetUserByIdQuery) {
    const { userId } = query;
    // 从读库（如 Redis / MongoDB / 只读从库）直接获取扁平化数据
    console.log(`正在从读库读取数据，用户ID: ${userId}`);
    return { id: userId, name: '张三', email: 'zhangsan@example.com' };
  }
}

Controller：分流读写请求

// user.controller.ts
import { Controller, Post, Get, Body, Param } from '@nestjs/common';
import { CommandBus, QueryBus } from '@nestjs/cqrs';
import { CreateUserCommand } from './create-user.command';
import { GetUserByIdQuery } from './get-user.query';

@Controller('users')
export class UserController {
  constructor(
    private readonly commandBus: CommandBus,
    private readonly queryBus: QueryBus,
  ) {}

  @Post() // 写操作
  async createUser(@Body() body: { name: string; email: string }) {
    return this.commandBus.execute(new CreateUserCommand(body.name, body.email));
  }

  @Get(':id') // 读操作
  async getUser(@Param('id') id: string) {
    return this.queryBus.execute(new GetUserByIdQuery(id));
  }
}

CRUD

Create Read Update Delete / 增查改删

Nest.js

https://www.npmjs.com/~nestjscore

https://www.npmjs.com/package/@nestjs/cqrs

refs

https://learn.microsoft.com/zh-cn/azure/architecture/patterns/cqrs

©xgqfrms 2012-2021

www.cnblogs.com/xgqfrms 发布文章使用：只允许注册用户才可以访问！

原创文章，版权所有©️xgqfrms, 禁止转载 🈲️，侵权必究⚠️！