@easyweb/活动
用于 Easyweb 微服务的共享 RabbitMQ 事件层 — 弹性连接管理器, 键入 出版商/订阅商基础类,以及保存事件名称的交换/路由-键注册 各项服务保持一致.
安装
amqplib 财务报告和审定财务报表 amqp-connection-manager 为: 同行依赖关系 ——消费服务拥有
因此,树上只有一个副本:
npm install @easyweb/events amqplib amqp-connection-manager
两份 amqp-connection-manager 意味着每个服务有两个TCP连接,两个
心跳计时器 和两个不协调的重联环 保持他们的最高水平.
连接
createRabbitMQConnection 每个进程拥有一个连接。 把它叫一次,在模块中,其余的
服务进口:
// src/lib/rabbitmq.ts
import { createRabbitMQConnection } from "@easyweb/events";
import config from "../config";
const { host, port, user, pass, vhost } = config.rabbitmq;
const rabbitmq = createRabbitMQConnection({
urls: [{ protocol: "amqp", hostname: host, port, username: user, password: pass, vhost }],
connectionName: "auth-service",
});
export const connectRabbitMQ = rabbitmq.connectRabbitMQ;
export const getChannel = rabbitmq.getChannel;
export const isRabbitMQConnected = rabbitmq.isConnected;
export const onRabbitMQReady = rabbitmq.onReady;
export const closeRabbitMQ = rabbitmq.close;
选项
| 选项 | 默认 | 页:1 |
|---|---|---|
urls | —— 说 | 直接传到 amqp-connection-manager字符串或 Options.Connect 对象. |
connectionName | 无 | 连接下的RabbitMQ管理UI中的表面. 设定它,这就是你如何区分服务. |
heartbeatIntervalInSeconds | 120 | |
onChannelSetup | 无 | 每(再)接通时重跑. 宣布交换,排队, 和绑在这里 所以地形学生存 频道娱乐. |
处理合同
| 方法 | 行为 |
|---|---|
connectRabbitMQ() | 解决一次,第一个成功连接. 绝不拒绝——但当经纪人倒台时,绝不解决. 别 await 它的前面 server.listen(),或者经纪人的出局 让你的服务离开它的港口。 开枪吧 void 后听后. |
getChannel() | 掷出 "RabbitMQ not initialized" 如果在第一个连接之前被调用. |
isConnected() | 目前状态. 对健康终点有用. |
onReady(cb) | 第一次连接时起火 在每次重接之后。如果已经连接,立即开火。 使用此选项来重组订阅者 . |
close() | 关闭频道然后连接 . |
重新连接使用有全焦的指数回放,封顶为30s,通过
findServers 钩. 紧张是故意的:它停止了所有重新连接的服务
在经纪人重启后.
出版
定义事件形状一次, 然后是 a Publisher 每个事件:
import { Publisher, Exchanges, RoutingKeys, UserCreatedEvent } from "@easyweb/events";
import { getChannel } from "../lib/rabbitmq";
class UserCreatedPublisher extends Publisher<UserCreatedEvent> {
exchange = Exchanges.AuthService;
routingKey = RoutingKeys.USER_CREATED;
}
await new UserCreatedPublisher(getChannel()).publish(
{ userId, name, email },
requestId, // optional — travels as the x-request-id header
);
publish (可延长, topic (默认情况下)发送持久信件,以及
如果频道缓冲器满了,则丢出.
签署
import { Subscriber, Exchanges, RoutingKeys, UserCreatedEvent } from "@easyweb/events";
import { onRabbitMQReady } from "../lib/rabbitmq";
class UserCreatedSubscriber extends Subscriber<UserCreatedEvent> {
exchange = Exchanges.AuthService;
routingKey = RoutingKeys.USER_CREATED;
async onMessage(data: UserCreatedEvent["data"], _msg, requestId: string) {
await createProfile(data.userId, requestId);
}
}
// Re-subscribe on every reconnect, not just at boot.
onRabbitMQReady((channel) => {
void new UserCreatedSubscriber(channel).subscribe();
});
subscribe() 给您,每个订阅者:
- 命名的持久队列
<exchange>.<routingKey>,通过protected queueName - 死信交换
<exchange>.dlx队列<queueName>.dead - 延迟队列
<queueName>.retry,没有消耗 prefetch(1)- 通过将信件停放在
<queueName>.retry增加额x-retry-count和期满EVENT_RETRY_DELAY_MS(默认2000年)。 它过期了, 空信输入<queueName>并再次受到审判——多达三次尝试,然后 消息被敲响并掉落 以阻止毒信息循环.
重试回到 这个 队列,永远不要到交换站。 五个服务捆绑
user.account.provisioned; 重排到交换机 重排 4 中的操作器
已经成功。 见
文件/平台/ADR-0005。 。 。 .
"Event handler failed" 登录于 warn 重试时仍保留 error 编辑
发送会丢弃信件。 信件字符串是固定的 — 一个 Loki 记录规则密钥
就在这.
requestId 从 x-request-id 信头,回到一个新鲜的UUID,这样每个
消息可以追踪.
事件登记
交换键和路由键都活在这个包中,所以没有服务发明自己的字符串.
enum Exchanges {
AuthService = "auth",
UserService = "user",
SubscriptionService = "subscription",
BillingService = "billing",
PaymentService = "payment",
}
enum RoutingKeys {
USER_CREATED = "auth.user.created",
USER_ACCOUNT_PROVISIONED = "user.account.provisioned",
USER_PROFILE_UPDATED = "user.profile.updated",
SUBSCRIPTION_PLAN_PUBLISHED = "subscription.plan.published",
BILLING_INVOICE_ISSUED = "billing.invoice.issued",
PAYMENT_CHARGE_SUCCEEDED = "payment.charge.succeeded",
}
运行密钥遵循 <service>.<entity>.<action>。添加一个事件意味着添加路径
键下的事件界面 src/events/types/,并发布新版本
软件包.
每个自有服务一个交换机, 只有该服务发布它. RoutingKeys 是一个
由他们共同拥有的单一平坦的enum,因此成员名会将所有者作为前缀-
USER_CREATED 属于 author 的 auth.user.created,而用户服务的密钥是 USER_PROFILE_*, (中文(简体) ),
USER_ADDRESS_* 诸如来.
PaymentService 还没有出版商; 付费服务尚未建立。 钥匙就在这里
因此,计费可以在编译时命名它们,所以当它降落时,字符串不能漂移.
电线上的钱
钱是小数串 小单位因为后面的柱子是BigInt
JSON.stringify 扔一个。 解析到 BigInt, 不是数字。 小单位取决于
货币,总在金额之外的一个字段中进行:
| 货币 | 次要单位 | "199000" 表示 |
|---|---|---|
| IDR (英语) | Rp 1(零小数) | 199.000卢比 |
| 美元 | 一分钱 | 1 990.00美元 (美元) |
IDR是零小数的 因为Xendit和Frede都在这里定居 携带赛恩会 指在收费前将每张发票的总额四舍五入到卢比亚,其余的 {\fn黑体\fs22\bord1\shad0\3ahbe\4ah00\fscx67\fscy66\2chffffff\3ch808080}必须存放在某个地方 {\fn黑体\fs22\bord1\shad0\3ahbe\4ah00\fscx67\fscy66\2chffffff\3ch808080}和解.
2.8.0的变化
subscription.subscription.activated 所得 所需经费 priceAmount, (中文(简体) ), currency 财务报告和审定财务报表
billingInterval。 。 。 .
对每一个消费者都有好处,对一个未成年人来说,而不是因为这个原因而成为主要消费者 实值
仅因 subscription.activated 完全没有
平台上一个出版商:订阅服务公司试版转换扫码.
他们曾经缺席过这个错误 Billing 订阅镜像部分合并
载荷,所以 activated 标出没有价格的行所持有的 subscription.created
当账户被提供时写作 free ——LifeTime间距为零价. 两者
计费周期的询问 完全排除了这种形状, 所以一个转换的试验是
从来没有计划,从来没有开过发票,从来没有收费, 顾客没有听说过这件事
因为通知对转换后的审理保持沉默,期望发票
为其说.
subscription.trial-started 刻意不携带,不得强迫携带.
Billing的策划者没有读过试验专栏,只有零价才留着
说出一个仍在审理中的审判.
2.0.0的突破变化
- IDR数量为零小数. 它们在1x是sen (exproduction 2),其中
"9900000"谓 99000卢比. subscription.subscription.{renewed,past-due,paused,resumed}移动到billing交换 作为billing.subscription.*所以只有账单知道 时间被支付和翻转, 或故障耗尽。 四个载荷丢失了entitlementsVersion——计费不能知道,订阅服务在计算时进行计算 应用事件——并获得invoiceId,因此每次过渡都可以追踪到 导致它的文件.- 剩下的九个
subscription.subscription.*获得的有效载荷version,则 订阅行的版本,因此镜像可以像其他复制行一样守护它们.
行版本
跨越服务边界的一行的每个有效载荷都带有 version从 1 开始
并且每一次公布的修改 都会提高1倍 这样一面镜子就能从
下一个 完全规则,和 VersionGuardOutcome 类型消费者应用
与,居住于 src/events/types/version-contract.ts。 。 。 .
这很重要,因为 subscribe() 通过将信件停放在延迟队列中来重试
让它回到过去 尾声 (见上文)
是正常操作,不是异常.
出版商不得推进 version 仅写未复制的一行。 A 类
登入时戳会使版本增高,从而给序列带来永久的缺口,以及
每一个消费者摊位背后.
一行的第一个版本由创建它的任何事件宣布—— user.account.provisioned
描述、偏好和通知行, user.address.created 给一个地址 所以
镜像从第一版开始,第一版的编辑以一至二的正版方式适用。 消费者应
仍然容忍一幅他们从未见过的一行的快照:出版是最佳的,
和创作事件 输给经纪人 退出 否则会拖住一行永远.
发展
npm install
npm run build # cleans dist, then tsc
npm pack --dry-run # verify tarball contents before publishing
npm publish