基于 Postgres 的通知模式（Notifier Pattern）

=# LISTEN test_topic;
LISTEN
Time: 2.828 ms

=# SELECT pg_notify('test_topic', 'test_message');
 pg_notify
-----------

(1 row)

Time: 17.892 ms
Asynchronous notification "test_topic" with payload "test_message" received from server process with PID 98481.

// Listen 返回一个订阅，调用者可以通过它接收通知通道的消息。
func (l *Notifier) Listen(channel string) *Subscription {
    l.mu.Lock()
    defer l.mu.Unlock()

    existingSubs := l.subscriptions[channel]

    sub := &Subscription{
        channel:        channel,
        listenChan:     make(chan string, 100), // 缓冲通道，用于存储通知消息
        notifyListener: l,
    }
    l.subscriptions[channel] = append(existingSubs, sub)

    if len(existingSubs) > 0 {
        // 如果该通道已有订阅，复用已建立的通道。
        sub.establishedChan = existingSubs[0].establishedChan
        sub.establishedChanClose = func() {} // 空操作，因为不是通道所有者
        return sub
    }

    // 通知器在成功建立 LISTEN 后关闭该通道，订阅者可据此确认。
    sub.establishedChan = make(chan struct{})
    sub.establishedChanClose = sync.OnceFunc(func() { close(sub.establishedChan) })

    l.channelChanges = append(l.channelChanges,
        channelChange{channel, sub.establishedChanClose, channelChangeOperationListen})

    // 取消 WaitForNotification 的阻塞以便立即处理通道变更。
    l.waitForNotificationCancel()

    return sub
}

// EstablishedC 是一个通道，当通知器成功建立连接后，该通道会被关闭。
// 这在测试用例中尤为有用，因为它可以用来确认监听器不仅已启动，
// 而且已成功在通道上开始监听后再继续操作。
// 如果是对已建立通道的新订阅，则 EstablishedC 已经关闭，因此等待它总是安全的。
//
// 无法完全保证通知器一定能成功建立监听，
// 因此调用方通常会结合 context 完成通道（done）、停止通道（stop channel）
// 和/或超时机制一起使用 `select`。
//
// 当通知器停止时，该通道会始终被关闭。
func (s *Subscription) EstablishedC() <-chan struct{} { return s.establishedChan }

func (l *Notifier) runOnce(ctx context.Context) error {
    // 处理订阅通道的变更，例如 `LISTEN` 或 `UNLISTEN` 操作。
    if err := l.processChannelChanges(ctx); err != nil {
        return err
    }

    // WaitForNotification 是一个阻塞函数。
    // 为了能够定期唤醒处理新的 `LISTEN` 或 `UNLISTEN` 操作，
    // 我们为监听操作设置了一个上下文超时（context deadline）。
    // 如果超时了，但原因与上下文超时无关，则不会将其视为错误。
    notification, err := func() (*pgconn.Notification, error) {
        const listenTimeout = 30 * time.Second

        // 创建一个带超时的上下文，用于控制等待通知的时限。
        ctx, cancel := context.WithTimeout(ctx, listenTimeout)
        defer cancel()

        // 提供一种机制，用于在有新的订阅变更时取消阻塞等待。
        l.mu.Lock()
        l.waitForNotificationCancel = cancel
        l.mu.Unlock()

        // 调用驱动层的 WaitForNotification 等待通知。
        notification, err := l.conn.WaitForNotification(ctx)
        if err != nil {
            return nil, xerrors.Errorf("等待通知时出错: %w", err)
        }

        return notification, nil
    }()
    if err != nil {
        // 如果错误是由于取消（cancellation）或超时引起，
        // 并且父上下文没有错误，则返回 nil。
        if (errors.Is(err, context.Canceled) ||
            errors.Is(err, context.DeadlineExceeded)) && ctx.Err() == nil {
            return nil
        }

        return err
    }

    // 在发送通知之前加读锁，确保多线程安全。
    l.mu.RLock()
    defer l.mu.RUnlock()

    // 通知订阅者（如果没有订阅者或订阅为空，则此操作无效）。
    for _, sub := range l.subscriptions[notification.Channel] {
        sub.listenChan <- notification.Payload
    }

    return nil
}

// 如果通知器出现不健康状态，则重启工作进程。
// 这种情况通常不会发生，因为通知器内置了一个重试机制，
// 在放弃之前会尽量保持连接状态。
notifier.AddUnhealthyCallback(closeShutdown)