# 事件驱动架构 ## 概述事件驱动架构 (EDA) 通过事件而不是直接调用进行通信，从而实现松散耦合、可扩展的系统。该技能涵盖消息队列、发布/订阅模式、事件源、CQRS、使用 sagas 的分布式事务管理以及使用 Kafka 的数据流。 ## 可用代理 * **高级软件工程师** (Opus) - 架构设计、模式选择、分布式系统设计 * **软件工程师**（十四行诗） - 事件处理程序实现，消费者/生产者代码 * **高级软件工程师** (Opus) - 事件安全、授权模式、消息加密 * **高级软件工程师** (Opus) - 消息代理设置、Kafka 集群、队列配置 ## 关键概念 ### 消息队列 **RabbitMQ 实现**：从“amqplib”导入amqp，{通道，连接}；接口队列配置{ 名称：字符串；耐用：布尔值； deadLetterExchange？：字符串； messageTtl？：号码；最大重试次数？：数量； } 类 RabbitMQClient { 私人连接：连接 |空=空；私人频道：频道 |空=空；异步连接（url：字符串）：Promise { this.connection = 等待 amqp.connect(url); this.channel = 等待this.connection.createChannel(); // 处理连接错误 this.connection.on("错误", (err) => { console.error("RabbitMQ 连接错误:", err); this.重新连接(url); }); } 异步 setupQueue(config: QueueConfig): Promise { if (!this.channel) throw new Error("未连接"); 常量选项： amqp.Options.AssertQueue = { 耐用：config.durable，参数：{}， }; 如果（config.deadLetterExchange）{ options.arguments!["x-dead-letter-exchange"] = config.deadLetterExchange; } 如果 (config.messageTtl) { options.arguments!["x-message-ttl"] = config.messageTtl; } 等待 this.channel.assertQueue(config.name, options); } 异步发布（队列：字符串，消息：未知，选项？：发布选项， ): 承诺 { if (!this.channel) throw new Error("未连接"); const content = Buffer.from(JSON.stringify(message)); constpublishOptions: amqp.Options.Publish = { 持久：真实， messageId: 选项?.messageId ||加密.randomUUID(), 时间戳：Date.now(), 标题：选项？.headers， }; this.channel.sendToQueue(队列、内容、publishOptions); } 异步消耗( 队列：字符串，处理程序：（消息：T，确认：() => 无效， nack: (重新排队？: 布尔值) => void, ) => 承诺, 选项？：ConsumOptions， ): 承诺 { if (!this.channel) throw new Error("未连接"); 等待 this.channel.prefetch(options?.prefetch || 10); 等待 this.channel.consume(queue, async (msg) => { if (!msg) 返回；尝试{ const 内容：T = JSON.parse(msg.content.toString()); 常量重试次数 = (msg.properties.headers?.["x-retry-count"] as number) || 0; 等待处理程序（内容， () => this.channel!.ack(msg), （重新排队=假）=> { if (重新排队 && 重试计数 < (选项?.maxRetries || 3)) { // 重新排队并增加重试次数 this.channel!.nack(msg, false, false); this.publish(队列, 内容, { headers: { "x-retry-count": retryCount + 1 }, }); } 否则{ this.channel!.nack(msg, false, false); // 发送到DLQ } }, ）； } 捕获（错误）{ console.error("消息处理错误：", error); this.channel!.nack(msg, false, false); } }); } } **AWS SQS 实施**：导入{ SQS客户端，发送消息命令，接收消息命令，删除消息命令，来自“@aws-sdk/client-sqs”；接口 SQSMessage { id：字符串；身体：T；收据句柄：字符串；近似接收计数：数量； } 类 SQSQueue { 私人客户端：SQSClient；私有队列Url：字符串；构造函数（queueUrl：字符串，区域：字符串=“us-east-1”）{ this.client = new SQSClient({ 区域 }); this.queueUrl =queueUrl; } 异步发送（消息：T，选项？：{delaySeconds？：数字； deduplicationId?: 字符串 }, ): 承诺<字符串> { const 命令 = new SendMessageCommand({ QueueUrl: this.queueUrl, 消息正文：JSON.stringify(消息), DelaySeconds：选项？.delaySeconds， MessageDeduplicationId：选项？.deduplicationId， MessageGroupId：选项？.deduplicationId ？ “默认”：未定义， }); const 响应 = 等待 this.client.send(command); 返回response.MessageId!; } 异步接收（最大消息数：数量 = 10， waitTimeSeconds：数字= 20， ): Promise[]> { const 命令 = new ReceiveMessageCommand({ QueueUrl: this.queueUrl, 最大消息数：最大消息数，等待时间秒：等待时间秒，属性名称：[“近似接收计数”]， }); const 响应 = 等待 this.client.send(command); return (response.Messages || []).map((msg) => ({ id: msg.MessageId!, body: JSON.parse(msg.Body!) as T, 收据句柄: msg.ReceiptHandle!, 近似接收计数：parseInt( msg.Attributes?.ApproximateReceiveCount || “1”, ), })); } 异步删除（receiptHandle：字符串）：Promise { 常量命令=新的DeleteMessageCommand({ QueueUrl: this.queueUrl, 收据句柄：收据句柄， }); 等待 this.client.send(command); } 异步进程消息（处理程序：（消息：T）=> Promise，选项？：{ maxRetries？：数量； pollInterval?: 数字 }, ): 承诺 { const maxRetries = 选项?.maxRetries || 3；而（真）{ const messages = wait this.receive(); 等待 Promise.all( messages.map（异步（消息）=> { 尝试{ 等待处理程序（msg.body）；等待 this.delete(msg.receiptHandle); } 捕获（错误）{ console.error(`处理消息 ${msg.id} 时出错：`, error); if (msg.approximateReceiveCount >= maxRetries) { // 可见性超时后消息将进入DLQ console.warn(`消息 ${msg.id} 超出最大重试次数`); } // 不要删除 - 将在可见性超时后重新处理 } }), ）； if (messages.length === 0 && 选项?.pollInterval) { 等待新的 Promise((r) => setTimeout(r, options.pollInterval)); } } } } ### 发布/订阅模式 **卡夫卡实施**：从“kafkajs”导入{ Kafka、生产者、消费者、EachMessagePayload }；接口事件 { id：字符串；类型：字符串；时间戳：日期；来源：字符串；数据：T；元数据？：记录<字符串，字符串>； } 类 KafkaEventBus { 私人卡夫卡：卡夫卡；私人制片人：制片人 |空=空；私人消费者：Map = new Map(); 构造函数（配置：{经纪人：字符串[]; clientId：字符串}）{ this.kafka = 新卡夫卡({ clientId: config.clientId, 经纪人：config.brokers， }); } 异步连接(): Promise { this.生产者 = this.kafka.生产者({ 幂等：正确，最大飞行请求数：5， }); 等待 this.生产者.connect(); } 异步发布( 主题：字符串，事件：省略 , "id" | “时间戳”>, ): 承诺 { if (!this. Producer) throw new Error("生产者未连接"); const fullEvent: 事件 = { ...事件， id: crypto.randomUUID(), 时间戳：新日期（）， }; 等待这个.生产者.发送({ 主题，消息：[ { 关键： event.data && typeof event.data === "object" && "id" in event.data ？ String((event.data as { id: 未知 }).id) ：完整事件.id，值：JSON.stringify(fullEvent), 标题：{ “事件类型”：事件类型， “事件源”：事件.源， }, }, ], }); } 异步订阅( 主题：字符串[]，组ID：字符串，处理程序：（事件：Event）=> Promise，选项？：{ fromBeginning？：布尔值}， ): 承诺 { const 消费者 = this.kafka.consumer({ groupId }); 等待消费者.connect(); for（主题的常量主题）{ 等待消费者.订阅({ 主题， fromBeginning: 选项?.fromBeginning, }); } this.consumers.set(groupId, 消费者); 等待消费者.run({ 每个消息：异步（{ 主题，分区，留言， }: 每个消息有效负载) => { 尝试{ const 事件：Event = JSON.parse(message.value!.toString()); 等待处理程序（事件）； } 捕获（错误）{ 控制台.错误（ `处理来自 ${topic}:${partition}: 的消息时出错：`, 错误，）；抛出错误； // 将根据消费者配置触发重试 } }, }); } 异步断开连接(): Promise { 等待 this.生产者？.disconnect(); for (constconsumer of this.consumers.values()) { 等待消费者.disconnect(); } } } // 用法 const eventBus = new KafkaEventBus({ 经纪人：[“本地主机：9092”]， clientId: "订单服务", }); 等待 eventBus.connect(); // 发布等待 eventBus.publish("订单", { 类型：“订单.创建”，来源：“订单服务”，数据：{ orderId：“123”，项目：[]，总计：99.99 }， }); // 订阅等待 eventBus.subscribe( [“订单”]， “库存服务”，异步（事件）=> { if (event.type === "order.created") { 等待 ReserveInventory(event.data); } }, ）； **NATS 实施**：导入{ 连接，国家连接，字符串编解码器，喷射流管理器， JetStream客户端， } 来自“nats”； NATSEventBus 类 { 私人 nc：NatsConnection |空=空；私有js：JetStreamClient |空=空；私有 sc = StringCodec(); 异步连接（服务器：字符串[]）：Promise { this.nc = 等待连接({ 服务器 }); // 设置 JetStream 以实现持久化 const jsm = 等待 this.nc.jetstreamManager(); this.js = this.nc.jetstream(); // 如果不存在则创建流尝试{ 等待 jsm.streams.add({ 名称：“活动”，主题：[“事件。*”]，保留：“限制”，最大消息数：1000000， max_age: 7 * 24 * 60 * 60 * 1000000000, // 7 天（纳秒） }); } 捕获 (e) { // 流可能已经存在 } } 异步发布（主题：字符串，数据：未知）：Promise { if (!this.js) throw new Error("未连接"); 等待 this.js.publish( `事件。${主题}`, this.sc.encode(JSON.stringify(data)), ）； } 异步订阅( 主题：字符串，持久名称：字符串，处理程序：（数据：未知）=> Promise， ): 承诺 { if (!this.js) throw new Error("未连接"); const 消费者 = 等待 this.js.consumers .get("事件", 持久名称) .catch(异步() => { // 如果不存在则创建消费者 const jsm = 等待 this.nc!.jetstreamManager(); 等待 jsm.consumers.add("活动", { 耐用名称：耐用名称， filter_subject: `事件。${subject}`, ack_policy: "明确的", 最大交付：3， }); return this.js!.consumers.get("EVENTS", DurableName); }); const messages =等待consumer.consume(); for wait (const msg of messages) { 尝试{ const data = JSON.parse(this.sc.decode(msg.data)); 等待处理程序（数据）；消息.ack(); } 捕获（错误）{ console.error("处理消息时出错：", error); 消息.nak(); } } } } ### 事件溯源接口域事件{ id：字符串；聚合 ID：字符串；聚合类型：字符串；类型：字符串；版本：编号；时间戳：日期；数据：未知；元数据：{ 用户ID？：字符串； correlationId？：字符串； causationId？：字符串； }; } 接口事件存储{ 追加（事件：DomainEvent []）：Promise ； getEvents(aggregateId: string, fromVersion?: number): Promise; getEventsByType(类型: string, fromTimestamp?: Date): Promise; } // PostgreSQL 事件存储类 PostgresEventStore 实现 EventStore { 构造函数（私有池：池）{} 异步追加（事件：DomainEvent []）：Promise { const client = wait this.pool.connect(); 尝试{ 等待客户端.query(“开始”); for（事件的常量事件）{ // 乐观并发检查 const { rows } = 等待 client.query( “从事件中选择 MAX（版本）作为 max_version，其中aggregate_id = $1”， [事件.aggregateId], ）； const currentVersion = rows[0]?.max_version || 0; if (event.version !== currentVersion + 1) { 抛出新的并发错误（ `预期版本 ${currentVersion + 1}，得到 ${event.version}`, ）； } 等待客户端.查询( `INSERT INTO 事件（id、aggregate_id、aggregate_type、类型、版本、时间戳、数据、元数据）值（$1、$2、$3、$4、$5、$6、$7、$8）`， [ 事件.id, 事件.aggregateId, 事件.aggregateType, 事件.类型, 事件版本，事件.时间戳, JSON.stringify(事件.数据), JSON.stringify(事件.元数据), ], ）； } 等待 client.query("COMMIT"); // 发布到事件总线以进行投影 for（事件的常量事件）{ 等待 this.eventBus.publish(event); } } 捕获（错误）{ 等待 client.query("ROLLBACK"); 抛出错误； }最后{ 客户端.release(); } } 异步 getEvents( 聚合 ID：字符串，来自版本：数字= 0， ): Promise { const { rows } = 等待 this.pool.query( `从事件中选择 * 其中aggregate_id = $1 并且版本> $2 按版本 ASC 排序`, [聚合 ID，来自版本]，）；返回 rows.map(this.rowToEvent); } } // 聚合基类抽象类聚合{ 私有_id：字符串；私人_版本：数字= 0；私有_uncommissedEvents：DomainEvent [] = []；获取 id(): 字符串 { 返回这个._id； } 获取版本（）：数字{ 返回这个._版本； } 构造函数（id：字符串）{ this._id = id; } 受保护的应用（事件：省略< 领域事件， “身份证” | “聚合ID”| “聚合类型”| “版本” | “时间戳” >, ): 无效 { const 域事件: 域事件 = { ...事件， id: crypto.randomUUID(), 聚合Id：这个._id，聚合类型: this.constructor.name, 版本：this._version + 1，时间戳：新日期（）， }; this.when(domainEvent); this._version = 域事件.version; this._uncommissedEvents.push(domainEvent); } 受保护的抽象当（事件：DomainEvent）：无效； loadFromHistory(事件：DomainEvent[]): void { for（事件的常量事件）{ this.when(事件); this._version = event.version; } } getUncommitedEvents(): DomainEvent[] { 返回[...this._uncommissedEvents]； } 清除未提交事件（）：无效{ this._uncommissedEvents = []; } } // 示例：订单聚合类订单扩展聚合{ 私人状态： | “待定” | “已确认” | “已发货” | “已交付” | “已取消”=“待处理”；私人物品：OrderItem[] = []；私人总数：数量= 0；静态创建（id：字符串，customerId：字符串，项目：OrderItem []）：订单{ const 订单 = 新订单(id); 订单.申请({ 类型：“订单已创建”，数据：{ 客户 ID，商品 }，元数据：{}， }); 退货订单； } 确认（）：无效{ if (this.status !== "待处理") { throw new Error("只能确认挂单"); } 这.应用（{ 输入：“订单已确认”，数据：{已确认：新日期（）}，元数据：{}， }); } 取消（原因：字符串）：void { if (["已发货", "已送达", "已取消"].includes(this.status)) { throw new Error("当前状态下无法取消订单"); } 这.应用（{ 类型：“订单已取消”，数据：{原因，cancelledAt：新日期（）}，元数据：{}， }); } 受保护的时间（事件：DomainEvent）：void { 开关（事件类型）{ 案例“订单创建”： const data = event.data as { items: OrderItem[] }; this.items = data.items; this.total = data.items.reduce( (总和, 项目) => 总和 + 项目.价格 * 项目.数量, 0, ）； this.status = "待处理"; 休息; 案例“订单已确认”： this.status = "已确认"; 休息; 案例“订单已取消”： this.status = "已取消"; 休息; } } } ### CQRS（命令查询职责分离） // 命令接口命令{ 类型：字符串；有效载荷：未知；元数据：{ 用户ID：字符串；关联 ID：字符串；时间戳：日期； }; } 接口 CommandHandler { 句柄（命令：T）：Promise ； } // 命令总线命令总线类 { 私有处理程序：Map> = new Map(); register(类型：字符串，处理程序：CommandHandler): void { this.handlers.set(type, handler as CommandHandler); } 异步调度（命令：命令）：Promise { const handler = this.handlers.get(command.type); 如果（！处理程序）{ throw new Error(`没有为命令注册处理程序：${command.type}`); } 等待处理程序.handle(命令); } } // 查询接口查询 { 类型：字符串；参数：未知； } 接口 QueryHandler, TResult> { 处理（查询：TQuery）：Promise ； } // 查询总线类查询总线{ 私有处理程序：Map,unknown>> = 新地图（）；注册, TResult>( 类型：字符串，处理程序：QueryHandler, ): 无效 { this.handlers.set(type, handler as QueryHandler,unknown>); } 异步执行 (查询: Query): Promise { const handler = this.handlers.get(query.type); 如果（！处理程序）{ throw new Error(`没有为查询注册处理程序：${query.type}`); } 返回 handler.handle(query) 作为 Promise; } } // 读取模型（投影）接口 OrderReadModel { id：字符串；客户 ID：字符串；客户名称：字符串；状态：字符串；项目：数组<{ 产品ID：字符串；产品名称：字符串；数量：数量；价格：数量； }>; 总计：数量；创建于：日期；更新时间：日期； } 类订单投影{ 构造函数（私有数据库：数据库，私有事件总线：EventBus，）{ this.setupSubscriptions(); } 私人 setupSubscriptions(): void { this.eventBus.subscribe("OrderCreated", this.onOrderCreated.bind(this)); this.eventBus.subscribe("订单确认", this.onOrderConfirmed.bind(this)); this.eventBus.subscribe("OrderCancelled", this.onOrderCancelled.bind(this)); } 私有异步 onOrderCreated(事件: DomainEvent): Promise { const data = event.data 作为 OrderCreatedData; // 丰富客户数据 const customer = wait this.db.customers.findById(data.customerId); // 丰富产品数据 const items = 等待 Promise.all( data.items.map（异步（项目）=> { const 产品 = 等待 this.db.products.findById(item.productId); 返回{ ...项目，产品名称：产品名称， }; }), ）；等待 this.db.orderReadModels.create({ id: 事件.aggregateId, 客户ID：数据.客户ID，客户名称：客户名称，状态：“待处理”，物品，总计： items.reduce((sum, i) => sum + i.price * i.quantity, 0), 创建时间：事件.时间戳，更新时间：事件.时间戳， }); } 私有异步 onOrderConfirmed(事件: DomainEvent): Promise { 等待 this.db.orderReadModels.update(event.aggregateId, { 状态：“已确认”，更新时间：事件.时间戳， }); } 私有异步 onOrderCancelled(事件: DomainEvent): Promise { 等待 this.db.orderReadModels.update(event.aggregateId, { 状态：“已取消”，更新时间：事件.时间戳， }); } } ### 分布式事务的 Saga 模式接口 SagaStep { 名称：字符串；执行：（数据：TData）=> Promise ；补偿：（数据：TData）=> Promise ； } 接口 SagaDefinition { 名称：字符串；步骤：SagaStep[]； } 接口 SagaInstance { id：字符串；传奇名称：字符串；数据：未知；当前步骤：编号；状态：“正在运行”| “已完成” | “补偿”| “失败的”; 已完成步骤：字符串[]；错误？：字符串；开始于：日期；更新时间：日期； } 类 SagaOrchestrator { 私有传奇：Map<字符串，SagaDefinition<未知>> = new Map(); 私人商店：SagaStore；寄存器(saga: SagaDefinition): void { this.sagas.set(saga.name, saga as SagaDefinition); } 异步开始 (sagaName: string, data: TData): Promise { const saga = this.sagas.get(sagaName); if (!saga) throw new Error(`未找到 Saga: ${sagaName}`); 常量实例：SagaInstance = { id: crypto.randomUUID(), 传奇名称, 数据，当前步数：0，状态：“正在运行”，已完成步骤: [], 开始于：新日期（），更新时间：新日期（）， }; 等待 this.store.save(instance); 等待 this.executeNextStep(实例, saga); 返回实例.id； } 私有异步执行NextStep( 实例：Saga实例，传奇：传奇定义<未知>， ): 承诺 { if (instance.currentStep >= saga.steps.length) { instance.status = "已完成"; 等待 this.store.save(instance); 返回; } const 步骤 = saga.steps[instance.currentStep]; 尝试{ 等待步骤.执行(实例.数据); instance.completedSteps.push(step.name); 实例.currentStep++; instance.updatedAt = new Date(); 等待 this.store.save(instance); 等待 this.executeNextStep(实例, saga); } 捕获（错误）{ instance.status = "正在补偿"; 实例.错误 = 错误实例错误？错误消息：字符串（错误）；等待 this.store.save(instance); 等待 this.compensate(实例, saga); } } 私有异步补偿（实例：Saga实例，传奇：传奇定义<未知>， ): 承诺 { // 按相反顺序执行补偿 for (让 i = instance.completedSteps.length - 1; i >= 0; i--) { const 步骤名称 = instance.completedSteps[i]; const step = saga.steps.find((s) => s.name === stepName); 如果（步骤）{ 尝试{ 等待步骤.补偿(实例.数据); } 捕获（错误）{ console.error(`步骤 ${stepName} 补偿失败:`, error); // 继续其他补偿 } } } 实例.status = "失败"; instance.updatedAt = new Date(); 等待 this.store.save(instance); } } // 示例：订单履行传奇接口订单履行数据{ 订单ID：字符串；客户 ID：字符串；项目：数组<{产品ID：字符串；数量：数量；价格：数量 }>; paymentId？：字符串；发货ID？：字符串； } const orderFulfillmentSaga: SagaDefinition = { 名称：“订单履行”，步骤：[ { name: "储备库存", 执行：异步（数据）=> { 等待 inventoryService.reserve(data.items); }, 补偿：异步（数据）=> { 等待 inventoryService.release(data.items); }, }, { name: “流程付款”, 执行：异步（数据）=> { const 总计 = data.items.reduce( (sum, i) => sum + i.价格 * i.数量， 0, ）； const payment = wait paymentService.charge(data.customerId, 总计); data. paymentId = 付款.id; }, 补偿：异步（数据）=> { if (data. paymentId) { 等待 paymentService.refund(data. paymentId); } }, }, { 名称：“创建发货”，执行：异步（数据）=> { 常量装运=等待shippingService.createShipment( 数据.orderId, 数据项，）； data.shipmentId = 发货.id; }, 补偿：异步（数据）=> { if (data.shipmentId) { 等待 ShippingService.cancelShipment(data.shipmentId); } }, }, { name: "确认订单", 执行：异步（数据）=> { 等待 orderService.confirm(data.orderId); }, 补偿：异步（数据）=> { 等待 orderService.cancel(data.orderId, "Saga 补偿"); }, }, ], }; ### 幂等性和一次性交付接口幂等性密钥{ 键：字符串；反应？：未知；创建于：日期；过期时间：日期； } 类幂等性服务 { 构造函数（私有redis：Redis）{} 异步进程( 键：字符串，操作： () => Promise, ttlSeconds: number = 86400, // 24 小时 ): 承诺 { const lockKey = `幂等性：锁：${key}`; const dataKey = `幂等性：数据：${key}`; // 尝试获取锁 const lock = wait this.redis.set(lockKey, "1", "EX", 30, "NX"); 如果（！锁定）{ // 另一个进程正在处理该请求，等待结果返回 this.waitForResult(dataKey); } 尝试{ // 检查是否已经处理 const现有=等待this.redis.get(dataKey); 如果（现有）{ 返回 JSON.parse(现有) 作为 T; } // 执行操作 const 结果 = 等待操作(); // 存储结果等待 this.redis.setex(dataKey, ttlSeconds, JSON.stringify(结果)); 返回结果； }最后{ 等待 this.redis.del(lockKey); } } 私有异步 waitForResult( 数据键：字符串， maxWaitMs：数量 = 30000， ): 承诺 { const startTime = Date.now(); while (Date.now() - startTime < maxWaitMs) { const data = wait this.redis.get(dataKey); 如果（数据）{ 返回 JSON.parse(data) 作为 T； } 等待新的 Promise((r) => setTimeout(r, 100)); } throw new Error("等待幂等操作结果超时"); } } // 为消费者进行消息去重类 DeduplicatingConsumer { 构造函数（私有redis：Redis， private windowSeconds: number = 3600, // 1 小时重复数据删除窗口）{} 异步进程（消息ID：字符串，处理程序：() => Promise, ): Promise<{ 结果: T;重复：布尔值 }> { const dedupKey = `dedup:${messageId}`; // 检查是否已经处理 const现有=等待this.redis.get(dedupKey); 如果（现有）{ return { 结果：JSON.parse(existing) as T，重复：true }; } // 处理消息 const 结果 = 等待处理程序(); // 标记为已处理等待 this.redis.setex( 重复数据删除密钥， this.windowSeconds, JSON.stringify(结果), ）；返回 { 结果，重复： false }; } } ### 死信队列接口死信消息{ id：字符串；原始队列：字符串；原始消息：未知；错误：字符串；失败于：日期；重试次数：次数；最后重试时间？：日期； } 类 DeadLetterQueueManager { 构造函数（私人 dlqStore：DLQStore，私有原始队列：消息队列，）{} 异步 moveToDeadLetter( 消息：未知，原始队列：字符串，错误：错误，重试次数：次数， ): 承诺 { const dlqMessage: DeadLetterMessage = { id: crypto.randomUUID(), 原始队列，原始消息：消息，错误：错误消息，失败处：新日期（），重试次数， }; 等待 this.dlqStore.save(dlqMessage); // DLQ 增长警报 const dlqSize = wait this.dlqStore.count(originalQueue); 如果（dlqSize > 100）{ 等待这个.alerting.warn({ title: "DLQ 大小警告", message: `${originalQueue} 的死信队列有 ${dlqSize} 消息`, }); } } 异步重试（messageId：字符串）：Promise { const dlqMessage = 等待 this.dlqStore.get(messageId); if (!dlqMessage) throw new Error("DLQ 中未找到消息"); 尝试{ 等待这个.originalQueue.publish( dlqMessage.originalQueue， dlqMessage.originalMessage, ）；等待 this.dlqStore.delete(messageId); } 捕获（错误）{ dlqMessage.lastRetryAt = new Date(); dlqMessage.retryCount++; 等待 this.dlqStore.save(dlqMessage); 抛出错误； } } async retryAll(queue: string): Promise<{ success: number;失败：数字 }> { const messages = wait this.dlqStore.getByQueue(queue); 让成功= 0；让失败= 0； for (const message of messages) { 尝试{ 等待 this.retry(message.id); 成功++； } 抓住 { 失败++； } } 返回{成功，失败}； } 异步清除（队列：字符串，oldThan？：日期）：Promise { 返回 this.dlqStore.deleteByQueue(queue, oldThan); } } ### 使用 Kafka 进行数据流传输 **流处理**：从“kafkajs”导入 { Kafka, CompressionTypes }；接口 StreamRecord { 键：字符串；值：T；时间戳：数字；分区：数量；偏移量：字符串； } 类 KafkaStreamProcessor { 私人卡夫卡：卡夫卡；构造函数（经纪人：字符串[]）{ this.kafka = 新卡夫卡({ clientId: "流处理器", 经纪人， }); } // 有状态流聚合异步聚合流( 输入主题：字符串，输出主题：字符串，组ID：字符串，初始状态：TState，聚合器：（状态：TState，记录：TInput）=> TState，窗口Ms：数量= 60000， ): 承诺 { const 消费者 = this.kafka.consumer({ groupId }); const 生产者 = this.kafka.生产者({ 压缩： CompressionTypes.GZIP， }); 等待消费者.connect(); 等待生产者.connect(); 等待消费者.subscribe({ 主题: inputTopic }); const stateByKey = new Map(); const windowTimers = new Map(); 等待消费者.run({ 每个消息： async ({ 消息 }) => { const key = message.key?.toString() || “默认”； const 值：TInput = JSON.parse(message.value!.toString()); // 获取或初始化状态 const currentState = stateByKey.get(key) ||初始状态； const newState = 聚合器(currentState, 值); stateByKey.set(key, newState); // 清除现有的窗口计时器 const existingTimer = windowTimers.get(key); 如果（现有定时器）clearTimeout（现有定时器）； // 设置新窗口计时器以发出聚合状态 const 计时器 = setTimeout(async () => { const FinalState = stateByKey.get(key); 等待生产者.发送（{ 主题：输出主题，消息：[ { 钥匙，值： JSON.stringify(finalState), 时间戳：Date.now().toString(), }, ], }); stateByKey.delete(key); windowTimers.delete(key); }, 窗口Ms); windowTimers.set(键, 计时器); }, }); } // 流连接异步 joinStreams( 左主题：字符串，右主题：字符串，输出主题：字符串，组ID：字符串，连接器：（左：TLeft，右：TRight）=> TResult，窗口Ms：数量= 30000， ): 承诺 { const 消费者 = this.kafka.consumer({ groupId }); const 生产者 = this.kafka. Producer(); 等待消费者.connect(); 等待生产者.connect(); 等待消费者.订阅({主题:[左主题,右主题]}); const leftCache = new Map(); const rightCache = new Map(); 等待消费者.run({ eachMessage: async ({ 主题, 消息 }) => { const key = message.key?.toString() || “默认”； const 时间戳 = parseInt(message.timestamp); const now = Date.now(); // 清理旧条目 this.cleanOldEntries(leftCache, now, windowMs); this.cleanOldEntries(rightCache, now, windowMs); 如果（主题===左主题）{ const leftData: TLeft = JSON.parse(message.value!.toString()); leftCache.set(key, { 数据: leftData, 时间戳 }); // 尝试用右连接 const rightEntry = rightCache.get(key); 如果（右输入&& Math.abs(时间戳 - rightEntry.timestamp) <= windowMs ）{ const 结果 = joiner(leftData, rightEntry.data); 等待生产者.发送（{ 主题：输出主题，消息：[{键，值：JSON.stringify（结果）}]， }); } } 否则{ const rightData: TRight = JSON.parse(message.value!.toString()); rightCache.set(key, { 数据: rightData, 时间戳 }); // 尝试用左连接 const leftEntry = leftCache.get(key); 如果（左条目 && Math.abs(时间戳 - leftEntry.timestamp) <= windowMs ）{ const 结果 = joiner(leftEntry.data, rightData); 等待生产者.发送（{ 主题：输出主题，消息：[{键，值：JSON.stringify（结果）}]， }); } } }, }); } 私有 cleanOldEntries( 缓存：Map<字符串，{ 数据：T；时间戳：数字 }>, 现在：数字，窗口Ms：数量， ): 无效 { for (cache.entries() 的 const [键，条目]) { if (now - Entry.timestamp > windowMs) { 缓存.删除(键); } } } } // Kafka Streams 风格的操作类 KafkaStream { 构造函数（私人卡夫卡：卡夫卡，私人主题：字符串，）{} // 地图变换映射（映射器：（值：T）=> R）：KafkaStream { const outputTopic = `${this.topic}-mapped`; this.processStream(outputTopic, 异步 (记录) => ({ key: 记录.key, 值：映射器（记录.值）， })); 返回新的 KafkaStream(this.kafka, outputTopic); } // 过滤器变换过滤器（谓词：（值：T）=>布尔值）：KafkaStream { const outputTopic = `${this.topic}-filtered`; this.processStream(outputTopic, 异步(记录) => 谓词(记录.值) ?记录：空，）；返回新的 KafkaStream(this.kafka, outputTopic); } // 按键分组并聚合 groupBy( keyExtractor：（值：T）=> K，聚合器：（键：K，值：T[]）=> V，窗口Ms：数量= 60000， ): KafkaStream { const outputTopic = `${this.topic}-分组`; const groups = new Map<字符串, T[]>(); const 计时器 = new Map(); this.processStream(outputTopic, async (记录, 生产者) => { const key = String(keyExtractor(record.value)); const 值 = groups.get(key) || []；值.push(记录.值); groups.set(键, 值); const现有定时器=timers.get(key); 如果（现有定时器）clearTimeout（现有定时器）； const 计时器 = setTimeout(async () => { const groupValues = groups.get(key) || []； const 结果 = 聚合器(keyExtractor(record.value), groupValues); 等待生产者.发送（{ 主题：输出主题，消息：[{键，值：JSON.stringify（结果）}]， }); groups.delete(key); 计时器.删除(键); }, 窗口Ms); 定时器.set（键，定时器）；返回空值； // 不要立即发出 }); 返回新的 KafkaStream(this.kafka, outputTopic); } 私有异步流程（输出主题：字符串，处理器：（记录：StreamRecord, 生产者：任何， ) => Promise<{ key: string;值：任意 } |空>, ): 承诺 { const 消费者 = this.kafka.consumer({ groupId: `${this.topic}-处理器`, }); const 生产者 = this.kafka. Producer(); 等待消费者.connect(); 等待生产者.connect(); 等待消费者.subscribe({ 主题: this.topic }); 等待消费者.run({ eachMessage: async ({ 消息, 分区 }) => { 常量记录：StreamRecord = { key: message.key?.toString() || “默认”，值： JSON.parse(message.value!.toString()), 时间戳：parseInt(消息.时间戳), 分区，偏移量：消息.偏移量， }; const 结果 = 等待处理器（记录，生产者）；如果（结果）{ 等待生产者.发送（{ 主题：输出主题，消息：[ { key: 结果.key, 值：JSON.stringify(结果.值), }, ], }); } }, }); } } ### 事件溯源模式 **性能快照**：界面快照{ 聚合 ID：字符串；版本：编号；状态：未知；时间戳：日期； } 类 SnapshotStore { 构造函数（私有数据库：数据库）{} 异步保存（快照：快照）：Promise { 等待 this.db.snapshots.upsert({ 聚合Id：快照.聚合Id，版本：快照.版本，状态：JSON.stringify(snapshot.state), 时间戳：快照.时间戳， }); } async getLatest(aggregateId: string): Promise<快照 |空> { const row = 等待 this.db.snapshots.findOne( { 聚合 ID }, { orderBy: { 版本: "desc" } }, ）；返回行？ { aggregateId: row.aggregateId, 版本：行.版本，状态：JSON.parse(row.state), 时间戳：行.时间戳， } ：空； } } // 使用快照增强聚合抽象类 SnapshotAggregate 扩展聚合 { 私有静态SNAPSHOT_FREQUENCY = 100； // 每 100 个事件快照异步加载（事件存储：事件存储，快照存储：快照存储， ): 承诺 { // 首先尝试从快照加载 const snapshot = wait snapshotStore.getLatest(this.id); 如果（快照）{ this.applySnapshot(snapshot.state); this._version = 快照.版本; // 加载快照以来的事件 const events = wait eventStore.getEvents(this.id, snapshot.version); this.loadFromHistory(事件); } 否则{ // 无快照，加载所有事件 const events = wait eventStore.getEvents(this.id); this.loadFromHistory(事件); } } 异步保存（事件存储：事件存储，快照存储：快照存储， ): 承诺 { const events = this.getUncommissedEvents(); 等待 eventStore.append(事件); this.clearUncommissedEvents(); // 如果需要创建快照 if (this.version % SnapshotAggregate.SNAPSHOT_FREQUENCY === 0) { 等待 snapshotStore.save({ 聚合Id：这个.id，版本：这个版本，状态：this.createSnapshot(), 时间戳：新日期（）， }); } } 受保护的抽象 createSnapshot()：未知；受保护的抽象 applySnapshot（状态：未知）：void； } **事件向上转换（架构迁移）**：接口EventUpcaster { 事件类型：字符串；来自版本：编号；版本：编号；向上转换：（事件：DomainEvent）=> DomainEvent； } 类 EventStoreWithUpcasting 实现 EventStore { 私有 upcasters: Map = new Map(); registerUpcaster(upcaster: EventUpcaster): void { const 现有 = this.upcasters.get(upcaster.eventType) || []；现有的.push(upcaster); 现有的.sort((a, b) => a.fromVersion - b.fromVersion); this.upcasters.set(upcaster.eventType, 现有); } 异步 getEvents( 聚合 ID：字符串，来自版本：数字= 0， ): Promise { const rawEvents = 等待 this.rawEventStore.getEvents( 聚合 ID, 来自版本，）； return rawEvents.map((event) => this.upcastEvent(event)); } 私人 upcastEvent(事件: DomainEvent): DomainEvent { const upcasters = this.upcasters.get(event.type) || []；让当前事件=事件； for (const upcaster of upcaster) { const eventVersion = (currentEvent.data as any)?.schemaVersion || 1; if (eventVersion === upcaster.fromVersion) { currentEvent = upcaster.upcast(currentEvent); } } 返回当前事件； } } // 向上转型者示例 const orderCreatedV1toV2: EventUpcaster = { 事件类型：“订单已创建”，来自版本：1，至版本：2，向上转换：（事件）=>（{ ...事件，数据：{ ...（事件数据任意）， // V2 添加了与账单分开的送货地址发货地址：（事件.数据为任意）.地址， billingAddress: (event.data 任意).address, 模式版本：2， }, }), }; ### 传奇模式 **编排与编排**： // 编排：服务独立地对事件做出反应订单服务类 { 异步 onOrderCreated(事件: OrderCreatedEvent): Promise { // 发布事件，其他服务做出反应等待 this.eventBus.publish("order.created", { orderId: 事件.orderId, customerId: 事件.customerId, 项目：事件.项目， }); } } 类库存服务{ 构造函数（私有 eventBus：EventBus）{ // 监听订单事件并做出反应 this.eventBus.subscribe("order.created", this.reserveInventory.bind(this)); } 私有异步reserveInventory（事件：OrderCreatedEvent）：Promise { 尝试{ 等待 this.reserve(event.items); // 发布成功事件等待 this.eventBus.publish("inventory.reserved", { orderId: 事件.orderId, }); } 捕获（错误）{ // 发布失败事件等待 this.eventBus.publish("inventory.reservation-failed", { orderId: 事件.orderId, 原因：错误消息， }); } } } 类支付服务{ 构造函数（私有 eventBus：EventBus）{ // 等待库存后再处理付款 this.eventBus.订阅（ "库存.预留", this.processPayment.bind(this), ）； } 私有异步 processPayment(事件: InventoryReservedEvent): Promise { // 处理付款并发布结果... } } // ORCHESTRATION：中央协调器控制流程 OrderFulfillmentOrchestrator 类 { 异步fullOrder(orderId: string): Promise { 尝试{ // 步骤1：储备库存等待 this.inventoryService.reserve(orderId); // 第 2 步：处理付款等待 this. paymentService.charge(orderId); // 第三步：创建发货等待 this.shippingService.createShipment(orderId); // 第四步：确认订单等待 this.orderService.confirm(orderId); } 捕获（错误）{ // 显式补偿等待 this.compensate(orderId); } } 私有异步补偿（orderId：字符串）：Promise { // 以相反的顺序撤消等待 this.shippingService.cancelShipment(orderId); 等待 this. paymentService.refund(orderId); 等待 this.inventoryService.release(orderId); 等待 this.orderService.cancel(orderId); } } **佐贺状态机**：类型 SagaState = | “开始” | “INVENTORY_RESERVED” | “PAYMENT_PROCESSED” | “已发货” | “完成” | “补偿” | “失败的”; 接口 SagaStateMachine { 状态：SagaState；数据：TData；转换：Map>； } 接口 SagaTransition { onEnter: (数据: TData) => Promise; onSuccess：SagaState； onFailure：SagaState；补偿？：（数据：TData）=> Promise ； } 类 StatefulSagaOrchestrator { 异步执行（saga：SagaStateMachine ）：Promise { 让 currentState = saga.state; while (currentState !== "COMPLETED" && currentState !== "FAILED") { const 转换 = saga.transitions.get(currentState); 如果（！过渡） throw new Error(`状态没有转换：${currentState}`); 尝试{ 等待transition.onEnter(saga.data); 当前状态 = 转换.onSuccess; saga.state = 当前状态; 等待 this.persistSaga(saga); // 保存状态 } 捕获（错误）{ // 补偿 if (transition.compensate) { 等待transition.compensate(saga.data); } 当前状态=transition.onFailure; saga.state = 当前状态; 等待 this.persistSaga(saga); } } } 私有异步 persistSaga(saga: SagaStateMachine): Promise { // 保存 saga 状态以供恢复等待 this.sagaStore.save({ id: (saga.data 任意).orderId, 状态：saga.state，数据：saga.data，更新时间：新日期（）， }); } } ## 最佳实践 1. **活动设计** * 事件应该是不可变的并且代表事实 * 使用过去时命名（OrderCreated，而不是CreateOrder） * 包括所有必要的数据；避免引用可变状态 * 版本化您的事件以进行模式演变 2. **幂等性** * 始终将消费者设计为幂等的 * 使用唯一的消息 ID 进行重复数据删除 * 存储处理状态以处理重试 3. **错误处理** * 为失败的消息实现死信队列 * 通过指数退避设置合理的重试限制 * 监控 DLQ 大小并对增长发出警报 4. **订购** * 在Kafka中使用分区键进行排序保证 * 理解至少一次与恰好一次的语义 * 需要时进行无序消息处理的设计 5. **监控** * 跟踪消息延迟、处理时间和错误率 * 设置消费者滞后警报 * 监控事件存储增长和查询性能 ## 示例 ### 完整的订单处理流程 // 1.API接收订单请求 app.post("/orders", async (req, res) => { 常量命令：CreateOrderCommand = { 类型：“创建订单”，有效负载：req.body，元数据：{ userId：req.user.id， correlationId: req.headers["x-correlation-id"] 作为字符串，时间戳：新日期（）， }, }; 等待 commandBus.dispatch(命令); res.status(202).json({ message: "订单创建已启动" }); }); // 2. 命令处理程序创建聚合并保存事件类 CreateOrderHandler 实现 CommandHandler { 异步句柄（命令：CreateOrderCommand）：Promise { 常量订单 = Order.create( 加密.randomUUID(), 命令.payload.customerId, 命令.payload.items, ）；等待 this.eventStore.append(order.getUncommissedEvents()); } } // 3. 事件发布到Kafka，预测更新读取模型 // 4. Saga Orchestrator 启动履行流程 // 5. 每个 saga 步骤都会发布更新投影的事件

事件驱动

包含技能