拓冰网站定制 
后端微服务云原生【免费下载链接】midway A Node.js Serverless Framework for front-end/full-stack developers. Build the application for next decade. Works on AWS, Alibaba Cloud, Tencent Cloud and traditional VM/Container. Super easy integrate with React and Vue. 项目地址https://gitcode.com/gh_mirrors/mi/midway点击查看免费下载本指南以 Midway 3.19.0 发布内容为核心详细介绍三大技术升级的落地方式全新重构的 Kafka 组件消费者/生产者/Admin 的多实例配置与 Kafka 实例复用、支持分组管理的 Mock 功能以及新增的perf-init启动性能分析命令。读者完成阅读后将能够在自己的 Midway 项目中迁移到新的 Kafka 配置体系、在复杂测试场景下精细控制 mock 数据的生命周期并为应用启动过程定位性能瓶颈。升级提示跨版本升级请参考 如何更新 Midway 中描述的步骤不要单独升级某个组件包以避免组件版本不一致导致的兼容性问题。一、3.19.0 版本更新总览v3.19 版本的主要变更集中在三个方面Kafka 组件重构配置和使用方法与之前有较大差异。虽然原有使用方式仍然兼容但文档不再保留旧写法官方文档全面切换到新配置体系。Mock 功能分组支持通过分组名称管理不同的 mock 数据可以单独恢复或清理某个分组的 mock。新增perf-init命令在开发环境下用于启动时初始化性能分析需要配合最新的mwtsc一同使用。此外版本还更新了一些依赖库以提高安全性和稳定性。二、Kafka 组件重构全新的配置与使用方式从 v3.19 开始Midway 的 Kafka 组件进行了重构配置和使用方法与之前有较大差异。重构后的组件基于kafkajs封装提供了更灵活的配置和更强大的功能支持消费者Consumer、生产者Producer和 Admin 三类角色三者可以独立或组合使用。2.1 新配置结构总览新版的 Kafka 配置统一收敛在kafka配置节点下分为consumer、producer、admin三个子节点// src/config/config.default.ts export default { kafka: { consumer: { sub1: { connectionOptions: { clientId: my-app, brokers: [localhost:9092], }, consumerOptions: { groupId: groupId-test-1, }, subscribeOptions: { topics: [topic-test-1], } }, }, producer: { clients: { pub1: { connectionOptions: { clientId: my-app, brokers: [localhost:9092], }, producerOptions: { // ... } } } } } }其中consumer下的每个键如sub1代表一个独立的消费者实例producer.clients和admin.clients下的每个键如pub1代表一个独立的生产者/Admin 实例采用服务工厂Service Factory设计模式管理。从源码结构看这一配置结构直接对应 packages/kafka/src/interface.ts 中定义的IMidwayKafkaConfigurationOptions其中consumer为命名实例集合producer与admin则通过ServiceFactoryConfigOption泛型约束为「clients 命名实例」结构。2.2 安装与开启组件安装依赖$ npm i midwayjs/kafka --savemidwayjs/kafka既可以作为独立主框架使用也可以附加在midwayjs/koa等其他主框架下// src/configuration.ts import { Configuration } from midwayjs/core; import * as koa from midwayjs/koa; import * as kafka from midwayjs/kafka; Configuration({ imports: [ koa, kafka ], // ... }) export class MainConfiguration { async onReady() { // ... } }从 packages/kafka/src/configuration.ts 的实现可以看到组件通过Configuration({ namespace: kafka })声明并在onReady阶段预加载KafkaProducerFactory同时默认注册了名为kafkaLogger的独立日志客户端输出到midway-kafka.log。2.3 消费者Consumer配置与实现2.3.1 多消费者配置通过consumer字段可以配置多个消费者sub1、sub2即两个不同的消费者。最简单的消费者配置需要三类字段连接配置connectionOptions、消费者配置consumerOptions和订阅配置subscribeOptions// src/config/config.default.ts export default { kafka: { consumer: { sub1: { connectionOptions: { clientId: my-app, brokers: [localhost:9092], }, consumerOptions: { groupId: groupId-test-1, }, subscribeOptions: { topics: [topic-test-1], } }, } } }完整可配置参数包括配置项说明底层对应connectionOptionsKafka 连接配置即new Kafka(connectionOptions)的参数如clientId、brokersKafkaConfigconsumerOptions消费者配置即kafka.consumer(consumerOptions)的参数如groupIdConsumerConfigsubscribeOptions订阅配置即consumer.subscribe(subscribeOptions)的参数如topics、fromBeginningConsumerSubscribeTopicsconsumerRunConfig消费者运行配置即consumer.run(consumerRunConfig)的参数如eachMessage/eachBatch之外的其他运行参数ConsumerRunConfig这些配置项的类型定义可参考 packages/kafka/src/interface.ts 中的IKafkaConsumerInitOptions其字段与 KafkaJS 原生 API 一一对应。2.3.2 复用 Kafka 实例kafkaInstanceRef多个消费者可以共享同一个 Kafka 实例通过kafkaInstanceRef字段指定// src/config/config.default.ts export default { kafka: { consumer: { sub1: { connectionOptions: { clientId: my-app, brokers: [localhost:9092], }, consumerOptions: { groupId: groupId-test-1, }, subscribeOptions: { topics: [topic-test-1], } }, sub2: { kafkaInstanceRef: sub1, consumerOptions: { groupId: groupId-test-2, }, subscribeOptions: { topics: [topic-test-2], } } } } }注意sub1和sub2是两个不同的消费者但共享同一个 Kafka 实例且sub2的groupId必须与sub1不同同一 groupId 下同一 topic 的分区会被均衡分配导致消息被两个消费者瓜分。用原生 KafkaJS 写法等价于const kafka new Kafka({ clientId: my-app, brokers: [localhost:9092], }); const consumer1 kafka.consumer({ groupId: groupId-test-1 }); const consumer2 kafka.consumer({ groupId: groupId-test-2 });kafkaInstanceRef的底层逻辑由 packages/kafka/src/framework.ts 与 packages/kafka/src/manager.ts 中的KafkaManager单例实现若指定了kafkaInstanceRef且已存在对应实例则直接复用否则创建新的Kafka实例并注册到实例映射中引用不存在的实例名会抛出kafka instance xxx not found错误。2.3.3 消费者实现类消费者实现类通过KafkaConsumer(sub1)装饰器与配置中的sub1消费者绑定实现eachMessage逐条消费// src/consumer/sub1.consumer.ts import { KafkaConsumer, IKafkaConsumer, EachMessagePayload } from midwayjs/kafka; KafkaConsumer(sub1) class Sub1Consumer implements IKafkaConsumer { async eachMessage(payload: EachMessagePayload) { // ... } }也可以实现eachBatch方法处理批量消息// src/consumer/sub1.consumer.ts import { KafkaConsumer, IKafkaConsumer, EachBatchPayload } from midwayjs/kafka; KafkaConsumer(sub1) class Sub1Consumer implements IKafkaConsumer { async eachBatch(payload: EachBatchPayload) { // ... } }从 packages/kafka/src/decorator.ts 的实现可以看到KafkaConsumer(consumerName)会保存模块元数据并将目标类注册为请求作用域RequestScope的Provide()类框架在运行时通过 packages/kafka/src/framework.ts 的resourceBinding流程根据实现类原型上是否存在eachBatch方法决定调用eachBatch还是eachMessage并把 KafkaJS 的run配置与业务方法绑定。2.3.4 消息上下文消息本身通过Context注入传递// src/consumer/sub1.consumer.ts import { KafkaConsumer, IKafkaConsumer, EachMessagePayload, Context } from midwayjs/kafka; import { Inject } from midwayjs/core; KafkaConsumer(sub1) class Sub1Consumer implements IKafkaConsumer { Inject() ctx: Context; async eachMessage(payload: EachMessagePayload) { // ... } }Context包含以下属性属性类型描述ctx.payloadEachMessagePayload/EachBatchPayload消息内容ctx.consumerConsumer消费者实例你可以通过ctx.consumer调用 KafkaJS 的 API例如ctx.consumer.commitOffsets手动提交偏移量或ctx.consumer.pause暂停消费。这一设计在 packages/kafka/src/interface.ts 中有完整定义其中topic、partition、message、commitOffsets等旧字段已标记为deprecated统一建议改用ctx.payload和ctx.consumer.commitOffsets。2.4 生产者Producer配置与使用生产者配置采用服务工厂设计模式通过producer.clients定义多个实例// src/config/config.default.ts export default { kafka: { producer: { clients: { pub1: { connectionOptions: { clientId: my-app, brokers: [localhost:9092], }, producerOptions: { // ... } } } } } }每个实例同样支持connectionOptionsnew Kafka()参数与producerOptionskafka.producer()参数。由于 Consumer 与 Producer 都可以从同一个 Kafka 实例创建生产者也可以复用消费者的 Kafka 实例// src/config/config.default.ts export default { kafka: { consumer: { sub1: { connectionOptions: { clientId: my-app, brokers: [localhost:9092], }, } }, producer: { clients: { pub1: { kafkaInstanceRef: sub1, } } } } }生产者不存在默认实例通过InjectClient()注入具名实例并使用// src/service/user.service.ts import { Provide, InjectClient } from midwayjs/core; import { KafkaProducerFactory, Producer } from midwayjs/kafka; Provide() export class UserService { InjectClient(KafkaProducerFactory, pub1) producer: Producer; async invoke() { await this.producer.send({ topic: topic-test-1, messages: [{ key: message-key1, value: hello consumer 11 ! }], }); } }KafkaProducerFactory的实现在 packages/kafka/src/service.ts 中它在Init()阶段通过initClients并发创建所有配置的生产者创建时支持kafkaInstanceRef实例复用并监听producer.connect事件输出kafkaLogger日志销毁时调用producer.disconnect()。值得关注的是该工厂内置了链路追踪tracing能力通过kafka.tracing.enable开关与kafka.tracing.injector自定义注入器在producer.send与sendBatch时自动把链路上下文注入到每条消息的headers中见 packages/kafka/src/service.ts方便跨服务串联调用链。2.5 Admin 配置与使用Kafka 的 Admin 功能用于创建、删除、查看主题查看配置和 ACL 等。Admin 同样使用服务工厂模式// src/config/config.default.ts export default { kafka: { admin: { clients: { admin1: { // ... } } } } }Admin 同样支持kafkaInstanceRef复用 Kafka 实例配置方式与 Producer 一致并通过InjectClient()注入// src/service/admin.service.ts import { Provide, InjectClient } from midwayjs/core; import { KafkaAdminFactory, Admin } from midwayjs/kafka; Provide() export class AdminService { InjectClient(KafkaAdminFactory, admin1) admin: Admin; }2.6 组件日志与 KafkaJS 模块Kafka 组件默认使用kafkaLogger日志记录在midway-kafka.log可通过配置修改// src/config/config.default.ts export default { midwayLogger: { clients: { kafkaLogger: { fileLogName: midway-kafka.log, }, }, }, }日志输出格式可单独配置export default { kafka: { // ... contextLoggerFormat: info { const { jobId, from } info.ctx; return ${info.timestamp} ${info.LEVEL} ${info.pid} ${info.message}; }, } }KafkaJS 模块可以通过KafkaJS字段获取import { KafkaJS } from midwayjs/kafka; const { ConfigResourceTypes } KafkaJS; // ...2.7 关于 KafkaJS v2 分区器的警告如果使用 KafkaJS v2.0.0启动时可能出现默认分区器变更的警告KafkaJS v2.0.0 switched default partitioner...。该警告源于 v2 默认启用了新分区器。若接受新行为但想关闭警告可设置环境变量KAFKAJS_NO_PARTITIONER_WARNING1或者显式声明分区器如需要保持与旧版本一致的分配行为使用LegacyPartitioner// src/config/config.default.ts import { KafkaJS } from midwayjs/kafka; const { Partitioners } KafkaJS; export default { kafka: { producer: { clients: { pub1: { // ... producerOptions: { createPartitioner: Partitioners.DefaultPartitioner, // ... createPartitioner: Partitioners.LegacyPartitioner, }, }, }, }, } }2.8 从测试用例看新配置的落地形态仓库中 packages/kafka/test/index.test.ts 的集成测试完整演示了新配置的落地通过KafkaConsumer(sub1)与KafkaConsumer(sub2)注册两个消费者sub2通过kafkaInstanceRef: sub1复用实例订阅不同 topic随后用createKafkaProducer创建生产者分别向两个 topic 发送消息验证两个消费者各收到一条消息expect(total).toEqual(2)。测试同时覆盖了消费者方法抛错、多 topic 订阅等场景是验证本机 Kafka 环境KAFKA_URL/KAFKA_BROKERS环境变量或localhost:9092下新组件行为的直接参考。三、Mock 功能分组精细化控制测试数据3.1 分组机制说明从3.19.0开始Midway 的 mock 功能支持通过分组来管理不同的 mock 数据。你可以在创建 mock 时指定一个分组名称这样可以在需要时单独恢复或清理某个分组的 mock 数据而无需影响其他分组的 mock。3.2 分组使用示例import { mockContext, restoreMocks } from midwayjs/mock; it(should test mock with groups, async () { const app await createApp(); // 创建普通对象的 mock const a {}; mockProperty(a, getUser, async () { return midway; }, group1); // 创建上下文的 mock mockContext(app, user, midway, group1); mockContext(app, role, admin, group2); // 恢复单个分组 restoreMocks(group1); // 恢复所有分组 restoreAllMocks(); });通过分组你可以更灵活地管理和控制 mock 数据特别是在复杂的测试场景中。3.3 分组 API 详解midwayjs/mock导出的相关 API实现在 packages/mock/src/mock.ts如下函数签名要点说明mockProperty(obj, key, value, group default)第 4 个参数为分组名mock 普通对象/实例的属性或方法mockClassProperty(clzz, propertyName, value, group default)作用于类的原型mock 类的属性或方法mockContext(app, key, value, group default)作用于应用上下文为ctx注入属性、getter/setter 或自定义回调mockSession(app, key, value, group default)作用于 sessionmock session 字段mockHeader(app, headerKey, headerValue, group default)作用于 headersmock 请求头restoreMocks(group default)恢复指定分组只清理该分组的所有 mockrestoreAllMocks()恢复所有分组清理全部 mock3.4 分组机制的底层实现从源码结构看分组能力由 packages/core/src/service/mockService.ts 中的MidwayMockService单例Scope(ScopeEnum.Singleton)实现分组存储内部维护mocks、contextMocks、cache三个Mapstring, ...键即为分组名mockProperty会把属性原始描述符Object.getOwnPropertyDescriptor与是否自有属性hasOwnProperty一起记录保证恢复时能精确还原见 mockService.ts。单分组恢复restore(group)调用restoreGroup逆序遍历该分组内的 mock 项原本非自有属性则delete自有属性则通过Object.defineProperty还原原始描述符同时清理该分组的上下文 mock 与缓存见 mockService.ts。全量恢复restoreAll()汇总所有分组的键逐一执行restoreGroup并清空模拟器列表见 mockService.ts。上下文注入applyContextMocks在每次请求上下文创建时遍历所有分组中与该app匹配的 mock 项通过属性描述符或回调函数注入ctx。3.5 分组行为的测试验证仓库中 packages/core/test/service/mockService.test.ts 的测试用例完整验证了分组语义分别用group1、group2mockctx.user与ctx.role后restore(group1)只让user恢复原状toBeUndefinedrole仍为admin对类属性分组 mock 后restore(group1)只还原invokegetName仍保持 mock 值restoreAll()后所有分组全部还原且isMocked(obj, key, group)返回false不传分组名时默认使用default分组restore()即可还原。这些用例表明分组机制天然支持「按测试用例/按模块划分 mock 作用域」在大型测试套件中可避免 mock 数据互相污染。四、新增启动性能分析命令 perf-init在开发环境下新增了一个perf-init命令用于在启动时初始化性能分析需要配合最新的mwtsc一同使用{ scripts: { dev: cross-env NODE_ENVlocal mwtsc --watch --run midwayjs/mock/app.js --perf-init } }该命令通过在启动命令中追加--perf-init标志让开发服务器在启动阶段即完成性能分析器的初始化从而可以在启动过程中就开始记录Node.js性能指标便于开发者观察应用初始化、依赖注入、模块加载等阶段的开销。从源码结构看Midway 在 packages/core/src/common/performanceManager.ts 中提供了MidwayPerformanceManager由 packages/core/src/index.ts 导出基于perf_hooks的performance.mark/performance.measure实现标记与测量并支持performance.clearMarks/clearMeasures清理。perf-init与mwtsc的组合正是为了让这一性能测量能力在应用启动的最早阶段编译与加载midwayjs/mock/app.js入口之前即可生效。适用前提perf-init面向开发环境需使用最新版本的mwtscTypeScript 编译与运行工具链旧版mwtsc可能无法识别该参数。五、升级注意事项整体升级勿单独升级组件请严格参考 如何更新 Midway 的步骤进行整体升级。Kafka 配置迁移如果你正在使用旧版 Kafka 组件v3.19 仍兼容旧用法但建议尽快迁移到本文所述的新配置体系consumer/producer.clients/admin.clientskafkaInstanceRef因为旧文档已不再保留后续维护与示例都将围绕新写法展开。依赖安全更新本次版本同步更新了部分依赖库以提升安全性与稳定性升级后建议完整回归测试。测试数据隔离使用分组 mock 后注意每个分组的恢复粒度——restoreMocks(groupName)只影响该分组restoreAllMocks()影响全部避免在测试收尾时误用 API 导致 mock 残留或过早还原。更多详细内容可参考 Kafka 组件文档 与仓库中的组件源码packages/kafka/src、packages/mock/src、packages/core/src/service/mockService.ts。赞分享后端微服务云原生【免费下载链接】midway A Node.js Serverless Framework for front-end/full-stack developers. Build the application for next decade. Works on AWS, Alibaba Cloud, Tencent Cloud and traditional VM/Container. Super easy integrate with React and Vue. 项目地址https://gitcode.com/gh_mirrors/mi/midway点击查看免费下载相关推荐Midway 集成 Apache Kafkamidwayjs/kafka 组件演进、架构与生产消费实践Midway 集成 Apache Kafkamidwayjs/kafka 组件演进、架构与生产消费实践 导读 本文以 midway 仓库中 midwayj后端微服务云原生Midway 集成 MikroORM 组件实战指南midwayjs/mikro 的核心能力与版本演进解析Midway 集成 MikroORM 组件实战指南midwayjs/mikro 的核心能力与版本演进解析 导读 midwayjs/mikro 是 Midw后端微服务云原生Commit Mono黑暗模式适配指南最佳字重选择和对比度优化Commit Mono黑暗模式适配指南最佳字重选择和对比度优化 Commit Mono是一款专注于提升阅读体验的匿名中性编程字体在黑暗模式下的显示效果直接影前端上一篇KeymouseGo终极指南如何轻松实现鼠标键盘自动化操作 下一篇QQ音乐解密工具QMCDecode三步解锁无损音乐自由播放创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Midway 3.19.0 版本实战指南：Kafka 组件重构、Mock 分组管理与启动性能分析
后端微服务云原生【免费下载链接】midway A Node.js Serverless Framework for front-end/full-stack developers. Build the application for next decade. Works on AWS, Alibaba Cloud, Tencent Cloud and traditional VM/Container. Super easy integrate with React and Vue. 项目地址https://gitcode.com/gh_mirrors/mi/midway点击查看免费下载本指南以 Midway 3.19.0 发布内容为核心详细介绍三大技术升级的落地方式全新重构的 Kafka 组件消费者/生产者/Admin 的多实例配置与 Kafka 实例复用、支持分组管理的 Mock 功能以及新增的perf-init启动性能分析命令。读者完成阅读后将能够在自己的 Midway 项目中迁移到新的 Kafka 配置体系、在复杂测试场景下精细控制 mock 数据的生命周期并为应用启动过程定位性能瓶颈。升级提示跨版本升级请参考 如何更新 Midway 中描述的步骤不要单独升级某个组件包以避免组件版本不一致导致的兼容性问题。一、3.19.0 版本更新总览v3.19 版本的主要变更集中在三个方面Kafka 组件重构配置和使用方法与之前有较大差异。虽然原有使用方式仍然兼容但文档不再保留旧写法官方文档全面切换到新配置体系。Mock 功能分组支持通过分组名称管理不同的 mock 数据可以单独恢复或清理某个分组的 mock。新增perf-init命令在开发环境下用于启动时初始化性能分析需要配合最新的mwtsc一同使用。此外版本还更新了一些依赖库以提高安全性和稳定性。二、Kafka 组件重构全新的配置与使用方式从 v3.19 开始Midway 的 Kafka 组件进行了重构配置和使用方法与之前有较大差异。重构后的组件基于kafkajs封装提供了更灵活的配置和更强大的功能支持消费者Consumer、生产者Producer和 Admin 三类角色三者可以独立或组合使用。2.1 新配置结构总览新版的 Kafka 配置统一收敛在kafka配置节点下分为consumer、producer、admin三个子节点// src/config/config.default.ts export default { kafka: { consumer: { sub1: { connectionOptions: { clientId: my-app, brokers: [localhost:9092], }, consumerOptions: { groupId: groupId-test-1, }, subscribeOptions: { topics: [topic-test-1], } }, }, producer: { clients: { pub1: { connectionOptions: { clientId: my-app, brokers: [localhost:9092], }, producerOptions: { // ... } } } } } }其中consumer下的每个键如sub1代表一个独立的消费者实例producer.clients和admin.clients下的每个键如pub1代表一个独立的生产者/Admin 实例采用服务工厂Service Factory设计模式管理。从源码结构看这一配置结构直接对应 packages/kafka/src/interface.ts 中定义的IMidwayKafkaConfigurationOptions其中consumer为命名实例集合producer与admin则通过ServiceFactoryConfigOption泛型约束为「clients 命名实例」结构。2.2 安装与开启组件安装依赖$ npm i midwayjs/kafka --savemidwayjs/kafka既可以作为独立主框架使用也可以附加在midwayjs/koa等其他主框架下// src/configuration.ts import { Configuration } from midwayjs/core; import * as koa from midwayjs/koa; import * as kafka from midwayjs/kafka; Configuration({ imports: [ koa, kafka ], // ... }) export class MainConfiguration { async onReady() { // ... } }从 packages/kafka/src/configuration.ts 的实现可以看到组件通过Configuration({ namespace: kafka })声明并在onReady阶段预加载KafkaProducerFactory同时默认注册了名为kafkaLogger的独立日志客户端输出到midway-kafka.log。2.3 消费者Consumer配置与实现2.3.1 多消费者配置通过consumer字段可以配置多个消费者sub1、sub2即两个不同的消费者。最简单的消费者配置需要三类字段连接配置connectionOptions、消费者配置consumerOptions和订阅配置subscribeOptions// src/config/config.default.ts export default { kafka: { consumer: { sub1: { connectionOptions: { clientId: my-app, brokers: [localhost:9092], }, consumerOptions: { groupId: groupId-test-1, }, subscribeOptions: { topics: [topic-test-1], } }, } } }完整可配置参数包括配置项说明底层对应connectionOptionsKafka 连接配置即new Kafka(connectionOptions)的参数如clientId、brokersKafkaConfigconsumerOptions消费者配置即kafka.consumer(consumerOptions)的参数如groupIdConsumerConfigsubscribeOptions订阅配置即consumer.subscribe(subscribeOptions)的参数如topics、fromBeginningConsumerSubscribeTopicsconsumerRunConfig消费者运行配置即consumer.run(consumerRunConfig)的参数如eachMessage/eachBatch之外的其他运行参数ConsumerRunConfig这些配置项的类型定义可参考 packages/kafka/src/interface.ts 中的IKafkaConsumerInitOptions其字段与 KafkaJS 原生 API 一一对应。2.3.2 复用 Kafka 实例kafkaInstanceRef多个消费者可以共享同一个 Kafka 实例通过kafkaInstanceRef字段指定// src/config/config.default.ts export default { kafka: { consumer: { sub1: { connectionOptions: { clientId: my-app, brokers: [localhost:9092], }, consumerOptions: { groupId: groupId-test-1, }, subscribeOptions: { topics: [topic-test-1], } }, sub2: { kafkaInstanceRef: sub1, consumerOptions: { groupId: groupId-test-2, }, subscribeOptions: { topics: [topic-test-2], } } } } }注意sub1和sub2是两个不同的消费者但共享同一个 Kafka 实例且sub2的groupId必须与sub1不同同一 groupId 下同一 topic 的分区会被均衡分配导致消息被两个消费者瓜分。用原生 KafkaJS 写法等价于const kafka new Kafka({ clientId: my-app, brokers: [localhost:9092], }); const consumer1 kafka.consumer({ groupId: groupId-test-1 }); const consumer2 kafka.consumer({ groupId: groupId-test-2 });kafkaInstanceRef的底层逻辑由 packages/kafka/src/framework.ts 与 packages/kafka/src/manager.ts 中的KafkaManager单例实现若指定了kafkaInstanceRef且已存在对应实例则直接复用否则创建新的Kafka实例并注册到实例映射中引用不存在的实例名会抛出kafka instance xxx not found错误。2.3.3 消费者实现类消费者实现类通过KafkaConsumer(sub1)装饰器与配置中的sub1消费者绑定实现eachMessage逐条消费// src/consumer/sub1.consumer.ts import { KafkaConsumer, IKafkaConsumer, EachMessagePayload } from midwayjs/kafka; KafkaConsumer(sub1) class Sub1Consumer implements IKafkaConsumer { async eachMessage(payload: EachMessagePayload) { // ... } }也可以实现eachBatch方法处理批量消息// src/consumer/sub1.consumer.ts import { KafkaConsumer, IKafkaConsumer, EachBatchPayload } from midwayjs/kafka; KafkaConsumer(sub1) class Sub1Consumer implements IKafkaConsumer { async eachBatch(payload: EachBatchPayload) { // ... } }从 packages/kafka/src/decorator.ts 的实现可以看到KafkaConsumer(consumerName)会保存模块元数据并将目标类注册为请求作用域RequestScope的Provide()类框架在运行时通过 packages/kafka/src/framework.ts 的resourceBinding流程根据实现类原型上是否存在eachBatch方法决定调用eachBatch还是eachMessage并把 KafkaJS 的run配置与业务方法绑定。2.3.4 消息上下文消息本身通过Context注入传递// src/consumer/sub1.consumer.ts import { KafkaConsumer, IKafkaConsumer, EachMessagePayload, Context } from midwayjs/kafka; import { Inject } from midwayjs/core; KafkaConsumer(sub1) class Sub1Consumer implements IKafkaConsumer { Inject() ctx: Context; async eachMessage(payload: EachMessagePayload) { // ... } }Context包含以下属性属性类型描述ctx.payloadEachMessagePayload/EachBatchPayload消息内容ctx.consumerConsumer消费者实例你可以通过ctx.consumer调用 KafkaJS 的 API例如ctx.consumer.commitOffsets手动提交偏移量或ctx.consumer.pause暂停消费。这一设计在 packages/kafka/src/interface.ts 中有完整定义其中topic、partition、message、commitOffsets等旧字段已标记为deprecated统一建议改用ctx.payload和ctx.consumer.commitOffsets。2.4 生产者Producer配置与使用生产者配置采用服务工厂设计模式通过producer.clients定义多个实例// src/config/config.default.ts export default { kafka: { producer: { clients: { pub1: { connectionOptions: { clientId: my-app, brokers: [localhost:9092], }, producerOptions: { // ... } } } } } }每个实例同样支持connectionOptionsnew Kafka()参数与producerOptionskafka.producer()参数。由于 Consumer 与 Producer 都可以从同一个 Kafka 实例创建生产者也可以复用消费者的 Kafka 实例// src/config/config.default.ts export default { kafka: { consumer: { sub1: { connectionOptions: { clientId: my-app, brokers: [localhost:9092], }, } }, producer: { clients: { pub1: { kafkaInstanceRef: sub1, } } } } }生产者不存在默认实例通过InjectClient()注入具名实例并使用// src/service/user.service.ts import { Provide, InjectClient } from midwayjs/core; import { KafkaProducerFactory, Producer } from midwayjs/kafka; Provide() export class UserService { InjectClient(KafkaProducerFactory, pub1) producer: Producer; async invoke() { await this.producer.send({ topic: topic-test-1, messages: [{ key: message-key1, value: hello consumer 11 ! }], }); } }KafkaProducerFactory的实现在 packages/kafka/src/service.ts 中它在Init()阶段通过initClients并发创建所有配置的生产者创建时支持kafkaInstanceRef实例复用并监听producer.connect事件输出kafkaLogger日志销毁时调用producer.disconnect()。值得关注的是该工厂内置了链路追踪tracing能力通过kafka.tracing.enable开关与kafka.tracing.injector自定义注入器在producer.send与sendBatch时自动把链路上下文注入到每条消息的headers中见 packages/kafka/src/service.ts方便跨服务串联调用链。2.5 Admin 配置与使用Kafka 的 Admin 功能用于创建、删除、查看主题查看配置和 ACL 等。Admin 同样使用服务工厂模式// src/config/config.default.ts export default { kafka: { admin: { clients: { admin1: { // ... } } } } }Admin 同样支持kafkaInstanceRef复用 Kafka 实例配置方式与 Producer 一致并通过InjectClient()注入// src/service/admin.service.ts import { Provide, InjectClient } from midwayjs/core; import { KafkaAdminFactory, Admin } from midwayjs/kafka; Provide() export class AdminService { InjectClient(KafkaAdminFactory, admin1) admin: Admin; }2.6 组件日志与 KafkaJS 模块Kafka 组件默认使用kafkaLogger日志记录在midway-kafka.log可通过配置修改// src/config/config.default.ts export default { midwayLogger: { clients: { kafkaLogger: { fileLogName: midway-kafka.log, }, }, }, }日志输出格式可单独配置export default { kafka: { // ... contextLoggerFormat: info { const { jobId, from } info.ctx; return ${info.timestamp} ${info.LEVEL} ${info.pid} ${info.message}; }, } }KafkaJS 模块可以通过KafkaJS字段获取import { KafkaJS } from midwayjs/kafka; const { ConfigResourceTypes } KafkaJS; // ...2.7 关于 KafkaJS v2 分区器的警告如果使用 KafkaJS v2.0.0启动时可能出现默认分区器变更的警告KafkaJS v2.0.0 switched default partitioner...。该警告源于 v2 默认启用了新分区器。若接受新行为但想关闭警告可设置环境变量KAFKAJS_NO_PARTITIONER_WARNING1或者显式声明分区器如需要保持与旧版本一致的分配行为使用LegacyPartitioner// src/config/config.default.ts import { KafkaJS } from midwayjs/kafka; const { Partitioners } KafkaJS; export default { kafka: { producer: { clients: { pub1: { // ... producerOptions: { createPartitioner: Partitioners.DefaultPartitioner, // ... createPartitioner: Partitioners.LegacyPartitioner, }, }, }, }, } }2.8 从测试用例看新配置的落地形态仓库中 packages/kafka/test/index.test.ts 的集成测试完整演示了新配置的落地通过KafkaConsumer(sub1)与KafkaConsumer(sub2)注册两个消费者sub2通过kafkaInstanceRef: sub1复用实例订阅不同 topic随后用createKafkaProducer创建生产者分别向两个 topic 发送消息验证两个消费者各收到一条消息expect(total).toEqual(2)。测试同时覆盖了消费者方法抛错、多 topic 订阅等场景是验证本机 Kafka 环境KAFKA_URL/KAFKA_BROKERS环境变量或localhost:9092下新组件行为的直接参考。三、Mock 功能分组精细化控制测试数据3.1 分组机制说明从3.19.0开始Midway 的 mock 功能支持通过分组来管理不同的 mock 数据。你可以在创建 mock 时指定一个分组名称这样可以在需要时单独恢复或清理某个分组的 mock 数据而无需影响其他分组的 mock。3.2 分组使用示例import { mockContext, restoreMocks } from midwayjs/mock; it(should test mock with groups, async () { const app await createApp(); // 创建普通对象的 mock const a {}; mockProperty(a, getUser, async () { return midway; }, group1); // 创建上下文的 mock mockContext(app, user, midway, group1); mockContext(app, role, admin, group2); // 恢复单个分组 restoreMocks(group1); // 恢复所有分组 restoreAllMocks(); });通过分组你可以更灵活地管理和控制 mock 数据特别是在复杂的测试场景中。3.3 分组 API 详解midwayjs/mock导出的相关 API实现在 packages/mock/src/mock.ts如下函数签名要点说明mockProperty(obj, key, value, group default)第 4 个参数为分组名mock 普通对象/实例的属性或方法mockClassProperty(clzz, propertyName, value, group default)作用于类的原型mock 类的属性或方法mockContext(app, key, value, group default)作用于应用上下文为ctx注入属性、getter/setter 或自定义回调mockSession(app, key, value, group default)作用于 sessionmock session 字段mockHeader(app, headerKey, headerValue, group default)作用于 headersmock 请求头restoreMocks(group default)恢复指定分组只清理该分组的所有 mockrestoreAllMocks()恢复所有分组清理全部 mock3.4 分组机制的底层实现从源码结构看分组能力由 packages/core/src/service/mockService.ts 中的MidwayMockService单例Scope(ScopeEnum.Singleton)实现分组存储内部维护mocks、contextMocks、cache三个Mapstring, ...键即为分组名mockProperty会把属性原始描述符Object.getOwnPropertyDescriptor与是否自有属性hasOwnProperty一起记录保证恢复时能精确还原见 mockService.ts。单分组恢复restore(group)调用restoreGroup逆序遍历该分组内的 mock 项原本非自有属性则delete自有属性则通过Object.defineProperty还原原始描述符同时清理该分组的上下文 mock 与缓存见 mockService.ts。全量恢复restoreAll()汇总所有分组的键逐一执行restoreGroup并清空模拟器列表见 mockService.ts。上下文注入applyContextMocks在每次请求上下文创建时遍历所有分组中与该app匹配的 mock 项通过属性描述符或回调函数注入ctx。3.5 分组行为的测试验证仓库中 packages/core/test/service/mockService.test.ts 的测试用例完整验证了分组语义分别用group1、group2mockctx.user与ctx.role后restore(group1)只让user恢复原状toBeUndefinedrole仍为admin对类属性分组 mock 后restore(group1)只还原invokegetName仍保持 mock 值restoreAll()后所有分组全部还原且isMocked(obj, key, group)返回false不传分组名时默认使用default分组restore()即可还原。这些用例表明分组机制天然支持「按测试用例/按模块划分 mock 作用域」在大型测试套件中可避免 mock 数据互相污染。四、新增启动性能分析命令 perf-init在开发环境下新增了一个perf-init命令用于在启动时初始化性能分析需要配合最新的mwtsc一同使用{ scripts: { dev: cross-env NODE_ENVlocal mwtsc --watch --run midwayjs/mock/app.js --perf-init } }该命令通过在启动命令中追加--perf-init标志让开发服务器在启动阶段即完成性能分析器的初始化从而可以在启动过程中就开始记录Node.js性能指标便于开发者观察应用初始化、依赖注入、模块加载等阶段的开销。从源码结构看Midway 在 packages/core/src/common/performanceManager.ts 中提供了MidwayPerformanceManager由 packages/core/src/index.ts 导出基于perf_hooks的performance.mark/performance.measure实现标记与测量并支持performance.clearMarks/clearMeasures清理。perf-init与mwtsc的组合正是为了让这一性能测量能力在应用启动的最早阶段编译与加载midwayjs/mock/app.js入口之前即可生效。适用前提perf-init面向开发环境需使用最新版本的mwtscTypeScript 编译与运行工具链旧版mwtsc可能无法识别该参数。五、升级注意事项整体升级勿单独升级组件请严格参考 如何更新 Midway 的步骤进行整体升级。Kafka 配置迁移如果你正在使用旧版 Kafka 组件v3.19 仍兼容旧用法但建议尽快迁移到本文所述的新配置体系consumer/producer.clients/admin.clientskafkaInstanceRef因为旧文档已不再保留后续维护与示例都将围绕新写法展开。依赖安全更新本次版本同步更新了部分依赖库以提升安全性与稳定性升级后建议完整回归测试。测试数据隔离使用分组 mock 后注意每个分组的恢复粒度——restoreMocks(groupName)只影响该分组restoreAllMocks()影响全部避免在测试收尾时误用 API 导致 mock 残留或过早还原。更多详细内容可参考 Kafka 组件文档 与仓库中的组件源码packages/kafka/src、packages/mock/src、packages/core/src/service/mockService.ts。赞分享后端微服务云原生【免费下载链接】midway A Node.js Serverless Framework for front-end/full-stack developers. Build the application for next decade. Works on AWS, Alibaba Cloud, Tencent Cloud and traditional VM/Container. Super easy integrate with React and Vue. 项目地址https://gitcode.com/gh_mirrors/mi/midway点击查看免费下载相关推荐Midway 集成 Apache Kafkamidwayjs/kafka 组件演进、架构与生产消费实践Midway 集成 Apache Kafkamidwayjs/kafka 组件演进、架构与生产消费实践 导读 本文以 midway 仓库中 midwayj后端微服务云原生Midway 集成 MikroORM 组件实战指南midwayjs/mikro 的核心能力与版本演进解析Midway 集成 MikroORM 组件实战指南midwayjs/mikro 的核心能力与版本演进解析 导读 midwayjs/mikro 是 Midw后端微服务云原生Commit Mono黑暗模式适配指南最佳字重选择和对比度优化Commit Mono黑暗模式适配指南最佳字重选择和对比度优化 Commit Mono是一款专注于提升阅读体验的匿名中性编程字体在黑暗模式下的显示效果直接影前端上一篇KeymouseGo终极指南如何轻松实现鼠标键盘自动化操作 下一篇QQ音乐解密工具QMCDecode三步解锁无损音乐自由播放创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考