各位观众，各位朋友，大家好！
今天咱们来聊聊一个听起来很高大上，但其实也没那么神秘的技术——CQRS，也就是Command Query Responsibility Segregation，中文名儿叫“命令查询职责分离”。说白了，就是读写分离架构。
CQRS：啥玩意儿？
想象一下，你是一家银行的柜员。你每天要做两件事：
- 处理业务 (Command)： 客户来存钱、取钱、转账，你负责修改账户信息。
- 查询信息 (Query)： 客户来查余额、查流水，你负责提供账户信息。
如果只有一个柜员，那他既要处理业务，又要查询信息，忙得焦头烂额。如果业务量一大，查个余额都得排队，效率低下。
CQRS就像是把这个柜员分成两个：一个专门负责处理业务 (Command)，一个专门负责查询信息 (Query)。
- Command (命令)： 负责修改系统状态，比如创建用户、更新订单等等。Command 通常不会返回数据，只返回操作是否成功。
- Query (查询)： 负责查询系统状态，比如获取用户信息、查询订单列表等等。Query 通常返回数据，不会修改任何状态。
为啥要用CQRS？
好，现在你可能会问，分工是好事儿，但为啥要这么复杂？直接一个数据库，增删改查不就得了？
是，简单粗暴确实可以解决问题。但随着系统复杂度增加，问题也会浮出水面：
- 性能瓶颈： 读写都在同一个数据库上，高并发读写容易导致数据库压力过大，性能下降。
- 数据模型不匹配： 写入需要的数据模型可能和查询需要的数据模型不一样。比如，写入时需要保存用户的详细信息，而查询时只需要显示用户的姓名和头像。
- 复杂查询： 有些复杂的查询需要大量的计算和聚合，直接在主数据库上执行会影响性能。
- 安全风险： 读写权限可能不一样。比如，普通用户只能查询自己的信息，而管理员可以查询所有用户的信息。
CQRS就能解决这些问题。它可以带来以下好处：
- 优化性能： 读写分离，可以使用不同的数据库或存储介质，针对读写进行优化。比如，读多写少的场景可以使用缓存。
- 简化数据模型： 可以针对读写使用不同的数据模型，提高灵活性和可维护性。
- 提高可伸缩性： 可以独立扩展读写服务，根据业务需求灵活调整。
- 增强安全性： 可以针对读写服务进行不同的权限控制。
CQRS的架构模式
CQRS不是银弹，它引入了更多的复杂性。下面我们来探讨一下CQRS的常见架构模式：
-
最简单的模式：读写分离数据库
这是最基础的CQRS实现。使用两个数据库：一个用于写入 (Command)，一个用于查询 (Query)。写入数据库的数据会同步到查询数据库。
- 优点： 简单易懂，容易实现。
- 缺点： 数据同步存在延迟，可能导致数据不一致。
# 示例：使用两个数据库 import sqlite3 # 写入数据库 write_db = sqlite3.connect('write.db') write_cursor = write_db.cursor() write_cursor.execute(''' CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY, name TEXT, email TEXT ) ''') write_db.commit() # 查询数据库 read_db = sqlite3.connect('read.db') read_cursor = read_db.cursor() read_cursor.execute(''' CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY, name TEXT, email TEXT ) ''') read_db.commit() # Command: 创建用户 def create_user(name, email): write_cursor.execute('INSERT INTO users (name, email) VALUES (?, ?)', (name, email)) write_db.commit() # 同步数据到查询数据库 (这里只是一个简单的示例，实际中需要更完善的同步机制) read_cursor.execute('INSERT INTO users (name, email) VALUES (?, ?)', (name, email)) read_db.commit() return True # Query: 获取用户列表 def get_users(): read_cursor.execute('SELECT id, name, email FROM users') return read_cursor.fetchall() # 使用示例 create_user('Alice', '[email protected]') users = get_users() print(users) write_db.close() read_db.close()这个例子里，
write.db负责写入，read.db负责查询。create_user函数会同时写入两个数据库。注意，这只是一个简单的示例，实际生产环境中，需要更可靠的数据同步机制，比如使用消息队列。 -
基于消息队列的CQRS
为了解决数据同步延迟的问题，可以使用消息队列。Command服务将修改数据的事件发布到消息队列，Query服务订阅消息队列，并更新查询数据库。
- 优点： 异步数据同步，降低延迟，提高系统响应速度。
- 缺点： 引入消息队列，增加了系统的复杂性。需要保证消息的可靠传递。
# 示例：使用RabbitMQ实现CQRS import pika import json import sqlite3 # 写入数据库 write_db = sqlite3.connect('write.db') write_cursor = write_db.cursor() write_cursor.execute(''' CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY, name TEXT, email TEXT ) ''') write_db.commit() # 查询数据库 read_db = sqlite3.connect('read.db') read_cursor = read_db.cursor() read_cursor.execute(''' CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY, name TEXT, email TEXT ) ''') read_db.commit() # RabbitMQ配置 rabbitmq_host = 'localhost' exchange_name = 'user_events' # Command: 创建用户 def create_user(name, email): write_cursor.execute('INSERT INTO users (name, email) VALUES (?, ?)', (name, email)) write_db.commit() # 发布消息到RabbitMQ connection = pika.BlockingConnection(pika.ConnectionParameters(host=rabbitmq_host)) channel = connection.channel() channel.exchange_declare(exchange=exchange_name, exchange_type='fanout') message = {'event_type': 'user_created', 'name': name, 'email': email} channel.basic_publish(exchange=exchange_name, routing_key='', body=json.dumps(message)) print(" [x] Sent %r" % message) connection.close() return True # Query: 获取用户列表 def get_users(): read_cursor.execute('SELECT id, name, email FROM users') return read_cursor.fetchall() # Query服务：订阅RabbitMQ消息 def consume_messages(): connection = pika.BlockingConnection(pika.ConnectionParameters(host=rabbitmq_host)) channel = connection.channel() channel.exchange_declare(exchange=exchange_name, exchange_type='fanout') result = channel.queue_declare(queue='', exclusive=True) queue_name = result.method.queue channel.queue_bind(exchange=exchange_name, queue=queue_name) def callback(ch, method, properties, body): message = json.loads(body.decode()) print(" [x] Received %r" % message) if message['event_type'] == 'user_created': read_cursor.execute('INSERT INTO users (name, email) VALUES (?, ?)', (message['name'], message['email'])) read_db.commit() channel.basic_consume(queue=queue_name, on_message_callback=callback, auto_ack=True) print(' [*] Waiting for messages. To exit press CTRL+C') channel.start_consuming() # 使用示例 # 首先启动Query服务（在另一个终端运行） # consume_messages() # 取消注释以运行Query服务 # 然后运行Command: # create_user('Bob', '[email protected]') # 最后查询： # users = get_users() # print(users) write_db.close() read_db.close()这个例子里，
create_user函数将用户创建事件发布到RabbitMQ。consume_messages函数（需要单独运行）订阅RabbitMQ，接收到消息后更新查询数据库。注意： 这个例子需要安装RabbitMQ，并且需要先启动Query服务，再运行Command。 记得把注释取消才能运行对应的部分。
-
事件溯源 (Event Sourcing) 与 CQRS
事件溯源是一种将系统状态的变化记录为一系列事件的模式。在CQRS中，可以将Command服务产生的事件存储到事件存储 (Event Store) 中，Query服务通过重放事件来构建查询数据库。
- 优点： 可以追溯系统状态的变化历史，方便审计和调试。可以将事件存储作为数据源，构建多种不同的查询数据库，满足不同的业务需求。
- 缺点： 实现复杂，需要考虑事件的版本管理和持久化。
事件溯源本身就是一个非常复杂的 topic，这里就不展开详细的代码示例了。 但是需要理解的是，事件溯源可以作为 CQRS 的一个补充，让你的系统拥有更强大的能力。
CQRS的优缺点总结
为了更清晰地了解CQRS，我们来总结一下它的优缺点：
| 优点 | 缺点 |
|---|---|
| 优化读写性能 | 引入复杂性，增加开发和维护成本 |
| 简化数据模型 | 需要处理数据同步问题，比如数据一致性、最终一致性等 |
| 提高可伸缩性 | 需要考虑事件的版本管理和持久化（如果使用事件溯源） |
| 增强安全性 | 并非所有系统都适合CQRS，需要根据实际情况进行评估 |
| 更好的适应微服务架构 | |
| 更容易实现领域驱动设计(DDD) |
什么时候应该使用CQRS？
CQRS不是万能的，它只适用于特定的场景。以下是一些适合使用CQRS的场景：
- 高并发读写： 系统需要处理大量的并发读写请求。
- 复杂查询： 系统需要执行复杂的查询，并且查询性能对用户体验至关重要。
- 不同的数据模型： 读写需要使用不同的数据模型。
- 需要审计和追溯： 系统需要记录数据的变化历史。
- 微服务架构： 在微服务架构中，每个服务可以独立选择是否使用CQRS。
什么时候不应该使用CQRS？
以下是一些不适合使用CQRS的场景：
- 简单的CRUD应用： 如果只是简单的增删改查，没有必要引入CQRS。
- 小型系统： CQRS会增加系统的复杂性，小型系统可能无法承受。
- 对数据一致性要求极高： CQRS的数据同步存在延迟，如果对数据一致性要求极高，需要慎重考虑。
CQRS的实现细节
在实际实现CQRS时，还需要考虑以下细节：
- 数据同步： 如何保证读写数据库的数据一致性？可以使用事务、消息队列、最终一致性等方法。
- 事件处理： 如何处理Command服务产生的事件？可以使用事件总线、消息队列等。
- 版本管理： 如何管理事件的版本？可以使用事件的版本号、事件的元数据等。
- 幂等性： 如何保证Command的幂等性？可以使用唯一ID、乐观锁等方法。
- 事务处理: Command的处理需要考虑事务性，保证数据的一致性。
代码示例：一个简单的CQRS实现
为了更具体地说明CQRS的实现，我们来看一个更完整的例子。这个例子使用Python和Flask框架，模拟一个简单的用户管理系统。
# app.py
from flask import Flask, request, jsonify
import uuid
import sqlite3
import threading
app = Flask(__name__)
# 数据库配置
DATABASE = 'user.db'
DATABASE_LOCK = threading.Lock()
def get_db():
with DATABASE_LOCK:
db = sqlite3.connect(DATABASE)
db.row_factory = sqlite3.Row # 返回字典形式的数据
return db
def init_db():
db = get_db()
with app.open_resource('schema.sql', mode='r') as f:
db.cursor().executescript(f.read())
db.commit()
db.close()
# 注册初始化数据库
with app.app_context():
init_db()
# Command Handler
class UserCommandHandler:
def create_user(self, name, email):
user_id = str(uuid.uuid4())
db = get_db()
try:
db.execute('INSERT INTO users (id, name, email) VALUES (?, ?, ?)',
(user_id, name, email))
db.commit()
return user_id
except Exception as e:
db.rollback()
raise e
finally:
db.close()
def update_user_email(self, user_id, new_email):
db = get_db()
try:
db.execute('UPDATE users SET email = ? WHERE id = ?', (new_email, user_id))
db.commit()
if db.total_changes == 0:
return False # 用户不存在
return True
except Exception as e:
db.rollback()
raise e
finally:
db.close()
# Query Handler
class UserQueryHandler:
def get_user(self, user_id):
db = get_db()
try:
cur = db.execute('SELECT id, name, email FROM users WHERE id = ?', (user_id,))
user = cur.fetchone()
return dict(user) if user else None
except Exception as e:
raise e
finally:
db.close()
def list_users(self):
db = get_db()
try:
cur = db.execute('SELECT id, name, email FROM users')
users = [dict(row) for row in cur.fetchall()]
return users
except Exception as e:
raise e
finally:
db.close()
# 初始化 Command 和 Query Handler
command_handler = UserCommandHandler()
query_handler = UserQueryHandler()
# API 接口
@app.route('/users', methods=['POST'])
def create_user():
data = request.get_json()
try:
user_id = command_handler.create_user(data['name'], data['email'])
return jsonify({'id': user_id}), 201
except Exception as e:
return jsonify({'error': str(e)}), 500
@app.route('/users/<user_id>', methods=['GET'])
def get_user(user_id):
user = query_handler.get_user(user_id)
if user:
return jsonify(user), 200
else:
return jsonify({'message': 'User not found'}), 404
@app.route('/users', methods=['GET'])
def list_users():
users = query_handler.list_users()
return jsonify(users), 200
@app.route('/users/<user_id>', methods=['PUT'])
def update_user_email(user_id):
data = request.get_json()
try:
updated = command_handler.update_user_email(user_id, data['email'])
if updated:
return jsonify({'message': 'User email updated successfully'}), 200
else:
return jsonify({'message': 'User not found'}), 404
except Exception as e:
return jsonify({'error': str(e)}), 500
if __name__ == '__main__':
app.run(debug=True)-- schema.sql
DROP TABLE IF EXISTS users;
CREATE TABLE users (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
email TEXT NOT NULL
);代码解释：
app.py： Flask 应用的主文件，定义了 API 接口和处理逻辑。schema.sql： 定义了数据库的表结构。UserCommandHandler： 负责处理 Command，包括创建用户和更新用户邮箱。UserQueryHandler： 负责处理 Query，包括获取用户和列出用户。- API 接口： 提供了创建用户、获取用户、列出用户和更新用户邮箱的接口。
如何运行：
- 确保安装了 Python 和 Flask (
pip install flask)。 - 运行
python app.py启动应用。 - 可以使用
curl或Postman等工具测试 API 接口。
这个例子只是一个简单的演示，实际生产环境中，需要考虑更多因素，比如数据验证、错误处理、安全性等等。
总结
CQRS是一个强大的架构模式，可以帮助我们构建高性能、可伸缩、易维护的系统。但是，它也引入了更多的复杂性，需要根据实际情况进行评估。希望今天的讲座能帮助大家更好地理解CQRS，并在合适的场景下应用它。
记住，技术是为业务服务的，不要为了用技术而用技术。选择最适合你的工具，才能事半功倍！
感谢大家的观看，下次再见！