好的，各位观众老爷们，晚上好！欢迎来到“代码界的相声大会”，今天咱们不聊风花雪月，聊点硬核的——Python事件源（Event Sourcing）。
开场白：啥是事件源？
话说，咱们写程序，尤其是数据库相关的，传统套路是“状态保存”。啥意思？就是你操作数据库，直接改数据，最后数据库里存的就是“最终状态”。
比如，你银行账户里有100块，取了20，数据库里就直接变成80了。之前的100块？没了，彻底没了，就像青春一样，一去不复返。
但是！人生不能重来，数据可以！事件源就是这么个思想：我不直接存最终状态，我存的是一系列发生的“事件”。
还是银行账户的例子，我不存最终的80块，我存的是：
- 初始存款：100
- 取款：20
要查你现在有多少钱？把这些事件“回放”一遍，算出来就行了。
事件源的优点，像数星星一样多
- 审计追踪： 每一笔操作都记录在案，谁偷了我的钱，一查就知道！
- 数据恢复： 就算数据库崩了，只要事件还在，就能重构出所有状态。
- 时间旅行： 想知道昨天下午3点你账户里有多少钱？回放到那个时间点就行。
- 解耦： 业务逻辑和数据存储分离，以后想换数据库？小菜一碟！
- 更好的性能： 对于某些场景，写入事件比更新状态更快。
事件源的缺点，也不是没有
- 复杂性： 实现起来比传统方式复杂，需要考虑事件存储、回放、版本控制等等。
- 最终一致性： 读取状态需要回放事件，可能会有延迟，不能保证实时一致。
- 事件演化： 事件的结构可能会改变，需要考虑如何兼容旧事件。
代码说话：一个简单的事件源示例
咱们用Python来模拟一个简单的银行账户，用事件源来管理账户余额。
import json
from datetime import datetime
class Event:
def __init__(self, event_type, data):
self.event_type = event_type
self.data = data
self.timestamp = datetime.utcnow().isoformat()
def to_dict(self):
return {
'event_type': self.event_type,
'data': self.data,
'timestamp': self.timestamp
}
def to_json(self):
return json.dumps(self.to_dict())
@classmethod
def from_dict(cls, event_dict):
return cls(event_dict['event_type'], event_dict['data'])
@classmethod
def from_json(cls, event_json):
event_dict = json.loads(event_json)
return cls.from_dict(event_dict)
class Account:
def __init__(self, account_id):
self.account_id = account_id
self.balance = 0
self.events = [] # 存储事件
def deposit(self, amount):
event = Event('deposit', {'amount': amount})
self.apply(event)
self.events.append(event)
def withdraw(self, amount):
if self.balance < amount:
raise ValueError("余额不足")
event = Event('withdraw', {'amount': amount})
self.apply(event)
self.events.append(event)
def apply(self, event):
if event.event_type == 'deposit':
self.balance += event.data['amount']
elif event.event_type == 'withdraw':
self.balance -= event.data['amount']
def get_balance(self):
return self.balance
def replay(self, events):
"""从事件列表中重构账户状态"""
self.balance = 0 # 重置余额
for event in events:
self.apply(event)
class EventStore:
def __init__(self, file_path):
self.file_path = file_path
def save_event(self, account_id, event):
with open(self.file_path, 'a') as f:
event_data = {'account_id': account_id, 'event': event.to_dict()}
f.write(json.dumps(event_data) + 'n')
def get_events(self, account_id):
events = []
try:
with open(self.file_path, 'r') as f:
for line in f:
event_data = json.loads(line)
if event_data['account_id'] == account_id:
events.append(Event.from_dict(event_data['event']))
except FileNotFoundError:
pass # 文件不存在，说明没有事件
return events
# 使用示例
account_id = "user123"
event_store = EventStore("events.txt") # 存储事件到文件
# 创建账户并存款
account = Account(account_id)
account.deposit(100)
event_store.save_event(account_id, account.events[-1]) # 保存事件
account.withdraw(20)
event_store.save_event(account_id, account.events[-1]) # 保存事件
print(f"当前余额: {account.get_balance()}") # 输出: 当前余额: 80
# 模拟账户重建
new_account = Account(account_id)
events = event_store.get_events(account_id)
new_account.replay(events) # 从事件重构状态
print(f"重建后的余额: {new_account.get_balance()}") # 输出: 重建后的余额: 80代码解释：
-
Event类： 定义了事件的结构，包括事件类型、数据和时间戳。to_dict和from_dict方法用于事件的序列化和反序列化。 -
Account类： 代表银行账户。deposit和withdraw方法分别代表存款和取款操作。apply方法根据事件类型更新账户余额。replay方法用于从事件列表重构账户状态。 -
EventStore类： 负责存储和读取事件。save_event方法将事件保存到文件。get_events方法从文件中读取指定账户的事件列表。
更高级的事件源：CQRS 来助阵
事件源经常和 CQRS (Command Query Responsibility Segregation，命令查询职责分离) 模式一起使用。
CQRS 就是把读和写操作彻底分开。
- Command (命令)： 用于修改状态的操作，比如存款、取款。
- Query (查询)： 用于读取状态的操作，比如查询余额。
CQRS 的好处是：
- 性能优化： 可以针对读和写操作分别进行优化。
- 可伸缩性： 可以独立扩展读和写服务。
- 复杂性管理： 可以更好地管理复杂的业务逻辑。
事件演化：新瓶装旧酒？没问题！
随着业务发展，事件的结构可能会发生变化。比如，原来 withdraw 事件只有 amount 字段，后来需要增加 transaction_id 字段。
这时候就需要考虑如何兼容旧事件。常见的做法是：
- 事件版本控制： 给每个事件加上版本号。
- 事件转换： 在回放事件时，将旧版本的事件转换为新版本。
# 事件版本控制示例
class WithdrawEventV1:
def __init__(self, amount):
self.amount = amount
self.version = 1
class WithdrawEventV2:
def __init__(self, amount, transaction_id):
self.amount = amount
self.transaction_id = transaction_id
self.version = 2
def handle_withdraw_event(event):
if event.version == 1:
# 处理V1版本的事件
print(f"V1版本取款: {event.amount}")
elif event.version == 2:
# 处理V2版本的事件
print(f"V2版本取款: {event.amount}, 交易ID: {event.transaction_id}")事件存储：选择困难症？不存在的！
事件可以存储在各种地方，比如：
- 关系型数据库： 比如 MySQL、PostgreSQL。
- NoSQL 数据库： 比如 MongoDB、Cassandra。
- 专门的事件存储数据库： 比如 EventStoreDB。
- 消息队列： 比如 Kafka、RabbitMQ。
选择哪种存储方式取决于你的具体需求，比如性能、可伸缩性、数据一致性等等。
总结：事件源，未来可期！
事件源是一种强大的设计模式，可以为你的系统带来很多好处。但是，它也增加了一定的复杂性，需要仔细考虑。
希望今天的“代码界的相声大会”能让你对事件源有一个初步的了解。记住，代码的世界没有绝对的对错，只有最适合你的选择。
最后，留几个思考题：
- 事件源适用于所有场景吗？哪些场景不适合使用事件源？
- 如何保证事件的顺序性？
- 如何处理并发的事件写入？
- 除了银行账户，还有哪些场景可以使用事件源？
各位观众老爷们，下课！