@cisstech/nestjs-pg-pubsub (英语)

它会做什么

PostgreSQL 触发器在您的表格中检测INSERT, UPDATE和DEELTEs. 在队列表格中坚持更改,然后拉出并发送到键入 NestJS 的收听者. 没有只投票的设计, 没有丢失的信息, 没有外部经纪人.

为什么这个存在

邮递SQL LISTEN/NOTIFY 对实时来说是巨大的,但并不可靠:通知是火灾和遗忘,在断开时丢失,而且有效载荷尺寸有限制。 这个库用耐久的队列把它包起来,这样你就能同时得到反应和保证的交付.

密钥属性

• 持久性: 通知前在 SQL 表格中持续更改,幸存的崩溃和重新启动
• 反应数字 : LISTEN/NOTIFY 启动即时处理,以倒计时计费器作为安全网
• 已订购:信息由ID处理,分批处理,分批处理 SELECT FOR UPDATE SKIP LOCKED
• 用后退重试: 失败的消息被重审, 以指数反转(2^{n}$分钟)
• 后压: 将同时提交的通知合并在一起,以便最多一次运行一个拉力周期
• 隔离池: 使用它自己的 pg.Pool,独立于 TypeORM ,以避免集合相争
• 零下限 DDL: 触发器是指纹( MD5) , 仅在配置更改时重新创建

安装

yarn add @cisstech/nestjs-pg-pubsub pg

支持NestJS v10+和v11+.

快速启动

// app.module.ts
@Module({
  imports: [
    TypeOrmModule.forRoot({
      /* ... */
    }),
    PgPubSubModule.forRoot({
      databaseUrl: process.env.DATABASE_URL,
    }),
  ],
  providers: [UserChangeListener],
})
export class AppModule {}

// user-change.listener.ts
@Injectable()
@RegisterPgTableChangeListener(User)
export class UserChangeListener implements PgTableChangeListener<User> {
  async process(changes: PgTableChanges<User>, ctx: PgTableChangeContext): Promise<void> {
    for (const insert of changes.INSERT) {
      console.log(`New user: ${insert.data.email}`)
    }

    for (const update of changes.UPDATE) {
      console.log(`Updated fields: ${update.data.updatedFields.join(', ')}`)
    }
  }
}

就是这样。 库自动创建触发器,队列表,并开始收听.

如何运作

1. PostgreSQL 表格更改时触发起火并插入一行 pg_pubsub_queue
2. 扳机发出一个 NOTIFY 带有频道名称
3. 库收到通知并从队列中取出待发信件
4. 信件发送到匹配 @RegisterPgTableChangeListener 类
5. 处理中的信件被标记为这样; 失败的信件被重审, 以指数回放
6. 一个背景传票人每隔60年运行一次 作为安全网 错过通知

重要制约因素

• 听者须快. 一个慢听者拖延了那张桌子的整个批次. 卸下重活到排队(Bull等),只是从收听者中排出.
• 使用 TransactionAdapter 用于交易性写作. 如果收听者需要在交易中写入 DB,请配置 transactionAdapter 并用它标记听众 @RegisterPgTableChangeListener(Entity, { transactional: true })。库将收听器的呼叫包入适配器,通过 ctx.transaction。没有适配器,则使用 ctx.onError 以信号失败.

配置

所有选项都是可选的,但 databaseUrl见 完整配置参考 详细情况.

密钥调制键 :

文档

完整文档: https://cisstech.github.io/nestkit/docs/nestjs-pg-pubsub/getting-started (中文(简体) )

许可证

麻省理工学院 马马杜·西塞