技术方案

为了实现的实时通知系统，实现了以下内容：

• Server-Sent Events (SSE) ：使用 SSE 在服务器和客户端之间建立实时、单向的通道，允许在通知发生时立即传递。
• Redis Pub/Sub： 使用 Redis pub/sub 作为消息系统，可以让多个发布者向多个订阅者发送消息。
• Notification Write API： 开发了一个通知写入 API 来收集来自其他应用程序的通知， 此 API 向 Redis 发送通知。
• Notification Read API： 也开发了通知读取API，订阅Redis，监听所有的通知。 收到通知后，读取 API 将它们发送到通过 SSE 连接到它的客户端。

为什么选择SSE

1.HTTP 轮询虽然是一个可行的选项，但不提供即时通知传递。 使用轮询，必须以特定时间间隔对服务器执行 ping 操作以检查是否有新事件，这可能会占用大量资源且效率低下，尤其是在许多用户订阅了通知的情况下。

2.另一方面，SSE 在服务器和客户端之间提供实时、单向的通道，允许在通知发生时立即传送，提供了实时性的保障。

3.虽然 WebSocket是实时通信的实现方案之一，但与 SSE 相比，它们需要更复杂的实现，虽然允许服务器和客户端之间进行双向通信，但由于只需要向客户端推送通知，因此选择 SSE 作为更简单、更高效的解决方案。

Redis 发布/订阅

为了从多个应用程序 pod 发送通知，使用了 Redis pub/sub。 Redis pub/sub 是一个消息系统，可以让多个发布者向多个订阅者发送消息。

import { Redis } from 'ioredis';

@Injectable()
export class RedisService implements OnModuleInit {

  ...

  async onModuleInit() {
    this.subscriber = new Redis({
      sentinels: this.redisSentinelConfig.addresses,
      name: this.redisSentinelConfig.masterName,
      password: this.redisSentinelConfig.password,
    });
    
    await this.subscriber.subscribe(this.redisSentinelConfig.channelName, async (err, count) => {
      if (err) {
        this.logger.error(`Failed to subscribe: ${err.message}`);
        return;
      }
      this.logger.log(`Subscribed successfully! This client is currently subscribed to ${count} channels`);
    });
    
    this.subscriber.on('message', async (channel, message: string) => {
      const liveNotification: LiveNotification = JSON.parse(message);
      await this.liveNotificationService.emit(liveNotification);
    });
  }
}

系统设计

为了收集来自其他应用程序的通知，开发了Notification Write API，此 API 向 Redis 发送通知。 Notification Read API订阅Redis，监听所有通知。 收到通知后，读取 API 将通过 SSE把消息发送到 连接的客户端。

由于多个读取 API 可能运行在不同的应用程序 Pod 中，因此每个读取 API 都会收到来自 Redis 的所有通知。Notification Read API将根据连接到当前 pod 的客户端过滤通知。 确保只向每个连接的客户端发送相关通知。

import { AuthGuard } from '@nestjs/passport';
import { Controller, Sse, UseGuards } from '@nestjs/common';

@Controller('/live-notification')
export class LiveNotificationController {
  constructor(private readonly liveNotificationService: LiveNotificationService) {}

  @Sse()
  @ApiBearerAuth()
  @UseGuards(AuthGuard('jwt'))
  public getEventsBySeller(@Tracers() tracers: ITracers, @SellerId() sellerId: number) {
    return this.liveNotificationService.subscribeForSeller(sellerId);
  }
}
 

import { EventEmitter } from 'events';
import { filter, fromEvent } from 'rxjs';

@Injectable()
export class LiveNotificationService implements OnModuleInit {
  
  private readonly emitter = new EventEmitter();
  
  ...
  
  public async emit(data: LiveNotification) {
    this.emitter.emit('liveNotification', { data });
  }
  
  public subscribeForSeller(sellerId: number) {
    const source = fromEvent(this.emitter, 'liveNotification');
    return source.pipe(
      filter(({ data: liveNotification }) => 
        liveNotification?.content == 'heartbeat' || 
        liveNotification?.sellerId == sellerId)
    );
  }
}
    

心跳消息

为避免服务器和浏览器之间的连接关闭，需要发送心跳消息来检测断开连接。为了解决这个问题，实现了每 30 秒发送一次心跳消息，从而解决了这个问题。

onModuleInit (): any { 
  setInterval ( () => { 
    const emitterListenerCount = this . emitter . listenerCount ( 'liveNotification' ); 
    this . logger . log ( `Heartbeat 消息发送到具有活动发射器侦听器计数的 SSE 客户端：${emitterListenerCount} ` ); 
    this . emitter . emit ( 'liveNotification' , { data : { content : 'heartbeat' } });
  }, 30000 ); 
}

性能表现

目前的通知系统有 90,000 个并发连接，在生产环境中运行 15 个 pod 来处理负载。每个 Kubernetes pod 消耗大约 800 MB 的内存和 300 Mi 的 CPU 资源。