如何在Fastify+Drizzle ORM中监听并处理PostgreSQL连接中断？

针对postgres-js无传统事件监听的问题，可通过以下方式实现连接中断监控与重连逻辑：

1. 利用postgres-js客户端错误回调 #

postgres-js构造选项提供onerror参数，可直接捕获连接池级错误，用于记录连接中断信息。

2. 实现自定义重连机制 #

检测到连接失效时，销毁旧客户端并重新初始化连接，同时更新Fastify实例上的mainDb装饰器，确保后续请求使用可用连接。

3. 统一查询错误处理 #

通过封装的初始化方法，自动处理连接验证与错误重试，避免业务代码重复处理连接异常。

修改后的完整插件代码 #

import type { PostgresJsDatabase } from 'drizzle-orm/postgres-js'
import type { FastifyInstance, FastifyPluginAsync } from 'fastify'
import { drizzle } from 'drizzle-orm/postgres-js'
import fp from 'fastify-plugin'
import postgres from 'postgres'
import { mainTables } from '.'

interface DrizzlePluginOptions {
  url: string
  poolSize: number
  retryDelay?: number // 重连间隔，默认3秒
}

declare module 'fastify' {
  interface FastifyInstance {
    mainDb: PostgresJsDatabase<typeof mainTables>
    _reinitializeMainDb: () => Promise<void> // 内部重连方法
  }
}

const mainDbPlugin: FastifyPluginAsync<DrizzlePluginOptions> = fp(
  async (server: FastifyInstance, opts: DrizzlePluginOptions) => {
    const retryDelay = opts.retryDelay || 3000
    let queryClient: ReturnType<typeof postgres>
    let db: PostgresJsDatabase<typeof mainTables>

    // 初始化/重连数据库核心方法
    const initializeDb = async () => {
      try {
        // 销毁旧连接（如果存在）
        if (queryClient) {
          await queryClient.end({ timeout: 5000 })
          server.log.info('旧数据库连接已关闭')
        }

        // 创建新客户端并绑定错误监听
        queryClient = postgres(opts.url, {
          max: opts.poolSize,
          onerror: (err) => {
            server.log.error({ err }, '检测到数据库连接错误')
            // 触发重连
            void server._reinitializeMainDb()
          },
          onnotice: (notice) => {
            server.log.warn({ notice }, '收到数据库通知')
          }
        })

        // 验证新连接可用性
        await queryClient`SELECT 1`
        db = drizzle(queryClient, { schema: mainTables })
        server.decorate('mainDb', db)
        server.log.info('数据库连接已重新建立')
      } catch (err) {
        server.log.error({ err }, '数据库初始化失败')
        // 延迟重试，避免频繁请求
        setTimeout(() => void server._reinitializeMainDb(), retryDelay)
      }
    }

    // 暴露重连方法到Fastify实例
    server.decorate('_reinitializeMainDb', async () => {
      await initializeDb()
    })

    // 首次初始化连接
    await initializeDb()

    server.addHook('onReady', async () => {
      server.log.info('主数据库插件已就绪')
    })

    server.addHook('onClose', async () => {
      if (queryClient) {
        await queryClient.end()
        server.log.info('服务器关闭时，数据库连接已清理')
      }
    })
  },
  {
    name: 'main-db-plugin',
  },
)

export default mainDbPlugin

关键说明 #

• 错误捕获：通过onerror回调监听连接池错误，一旦触发立即记录日志并启动重连流程。
• 自动重连：_reinitializeMainDb方法封装了连接销毁、重建、验证的完整逻辑，失败后自动延迟重试，避免资源耗尽。
• 状态同步：每次重连后更新mainDb装饰器，确保Fastify实例上的数据库客户端始终可用。
• 优雅关闭：在Fastify的onClose钩子中销毁连接池，避免服务器 shutdown 时出现资源泄漏。

内容的提问来源于stack exchange，提问作者Nazar Duma