如何用.NET Core Web API的SSE实现Angular分数组件自动刷新

• 开发Angular端的队伍分数展示组件，当后端TblVottings表的分数数据发生变更时，前端无需用户操作即可自动同步刷新展示内容
• 采用SSE（服务端事件推送）方案实现自动同步逻辑，在现有.NET Core Web API的PUT接口完成数据更新后，主动触发前端的分数刷新动作
• 基于现有已实现的TblVottings控制器（含全量分数查询GET接口、会话投票数据更新PUT接口）完成改造，不重复编写已有业务逻辑

1. 实现SSE广播服务 #

新建单例服务统一管理前端的SSE长连接，在数据更新时给所有连接的客户端推送通知：

public class SseScoreBroadcaster
{
    // 存储所有活跃的前端连接
    private readonly ConcurrentDictionary<Guid, Func<string, Task>> _activeConnections = new();

    // 注册新的前端连接，返回唯一连接ID
    public Guid RegisterConnection(Func<string, Task> onPushMessage)
    {
        var connectionId = Guid.NewGuid();
        _activeConnections.TryAdd(connectionId, onPushMessage);
        return connectionId;
    }

    // 移除已断开的无效连接
    public void RemoveConnection(Guid connectionId)
    {
        _activeConnections.TryRemove(connectionId, out _);
    }

    // 给所有活跃连接推送分数更新通知
    public async Task BroadcastScoreUpdate()
    {
        foreach (var connection in _activeConnections.Values)
        {
            await connection("score_refresh_required");
        }
    }
}

在项目启动配置中注册该单例服务：

// Program.cs
builder.Services.AddSingleton<SseScoreBroadcaster>();

2. 改造现有TblVottings控制器 #

在保留原有GET、PUT接口核心逻辑的基础上，新增SSE连接端点，在PUT接口更新成功后触发广播：

[ApiController]
[Route("api/TblVottings")]
public class TblVottingsController : ControllerBase
{
    private readonly VotingContext _context;
    private readonly SseScoreBroadcaster _broadcastService;

    // 依赖注入数据库上下文和广播服务
    public TblVottingsController(VotingContext context, SseScoreBroadcaster broadcastService)
    {
        _context = context;
        _broadcastService = broadcastService;
    }

    // 原有全量分数查询GET接口完全保留
    [HttpGet]
    public async Task<ActionResult<IEnumerable<TblVotting>>> GetTblVottings()
    {
        return await _context.TblVottings.ToListAsync();
    }

    // 新增SSE长连接端点，供前端建立监听
    [HttpGet("score-stream")]
    public async Task ScoreEventStream()
    {
        // 设置SSE标准响应头
        Response.ContentType = "text/event-stream";
        Response.Headers["Cache-Control"] = "no-cache";

        // 注册当前前端连接
        var connectionId = _broadcastService.RegisterConnection(async message =>
        {
            // 按照SSE格式写入消息内容
            await Response.WriteAsync($"data: {message}\n\n");
            await Response.Body.FlushAsync();
        });

        // 前端断开连接时自动清理连接记录
        HttpContext.RequestAborted.Register(() => _broadcastService.RemoveConnection(connectionId));
        // 保持长连接直到客户端主动断开
        await Task.Delay(Timeout.Infinite, HttpContext.RequestAborted);
    }

    // 改造原有PUT更新接口
    [HttpPut("{id}")]
    public async Task<IActionResult> PutTblVotting(int id, TblVotting tblVotting)
    {
        if (id != tblVotting.SessionId)
        {
            return BadRequest();
        }

        _context.Entry(tblVotting).State = EntityState.Modified;

        try
        {
            await _context.SaveChangesAsync();
            // 数据保存成功后，给所有前端连接推送更新通知
            await _broadcastService.BroadcastScoreUpdate();
        }
        catch (DbUpdateConcurrencyException)
        {
            if (!TblVottingExists(id))
            {
                return NotFound();
            }
            else
            {
                throw;
            }
        }

        return NoContent();
    }

    private bool TblVottingExists(int id)
    {
        return _context.TblVottings.Any(e => e.SessionId == id);
    }
}

1. 封装SSE连接服务 #

新建根服务统一管理SSE连接，处理断线重连、Angular变更检测适配逻辑：

import { Injectable, NgZone } from '@angular/core';
import { Observable } from 'rxjs';

@Injectable({ providedIn: 'root' })
export class ScoreStreamService {
  private eventSourceRef!: EventSource;

  constructor(private ngZone: NgZone) {}

  // 订阅分数更新事件的Observable
  subscribeToScoreUpdates(): Observable<string> {
    return new Observable(observer => {
      // 连接后端的SSE端点
      this.eventSourceRef = new EventSource('/api/TblVottings/score-stream');

      // 处理收到的推送消息
      this.eventSourceRef.onmessage = (event) => {
        // 用NgZone包裹回调，确保Angular能触发视图变更检测
        this.ngZone.run(() => {
          observer.next(event.data);
        });
      };

      // 处理连接异常，实现自动重连
      this.eventSourceRef.onerror = (err) => {
        observer.error(err);
        this.eventSourceRef.close();
        // 3秒后自动重连
        setTimeout(() => this.subscribeToScoreUpdates().subscribe(), 3000);
      };

      // 组件销毁时的清理逻辑，关闭连接避免内存泄漏
      return () => this.eventSourceRef.close();
    });
  }
}

2. 实现分数展示组件 #

在组件初始化时建立SSE监听，收到更新通知后拉取最新分数渲染：

import { Component, OnDestroy, OnInit } from '@angular/core';
import { Subscription } from 'rxjs';
import { ScoreStreamService } from './score-stream.service';
import { HttpClient } from '@angular/common/http';
import { TblVotting } from './tbl-votting.model';

@Component({
  selector: 'app-score-board',
  template: `
    <div class="score-wrapper">
      <div class="score-item" *ngFor="let item of scoreList">
        <span class="team-name">{{ item.teamName }}</span>
        <span class="team-score">{{ item.score }}</span>
      </div>
    </div>
  `
})
export class ScoreBoardComponent implements OnInit, OnDestroy {
  scoreList: TblVotting[] = [];
  private streamSubscription!: Subscription;

  constructor(
    private scoreStream: ScoreStreamService,
    private http: HttpClient
  ) {}

  ngOnInit(): void {
    // 页面首次加载时拉取全量分数
    this.refreshScoreList();
    // 建立SSE监听，收到通知就刷新分数
    this.streamSubscription = this.scoreStream
      .subscribeToScoreUpdates()
      .subscribe(() => this.refreshScoreList());
  }

  // 封装拉取最新分数的请求方法
  private refreshScoreList() {
    this.http.get<TblVotting[]>('/api/TblVottings').subscribe(data => {
      this.scoreList = data;
    });
  }

  ngOnDestroy(): void {
    // 组件销毁时退订SSE监听
    this.streamSubscription.unsubscribe();
  }
}

• 跨域适配：如果前后端部署在不同域名/端口，需在后端CORS配置中允许对应前端源的SSE长连接，避免浏览器拦截事件流
• 性能优化：当前实现为“收到更新通知后重新拉取全量分数”，如果单表分数数据量较大，可直接在SSE推送的消息体中携带变更后的具体分数数据，前端收到后直接更新本地列表，减少额外HTTP请求开销
• 多实例适配：当前示例使用内存字典存储客户端连接，适合单实例部署场景；如果后续横向扩容为多实例部署，需将内存广播替换为分布式消息总线，保证所有实例上的连接客户端都能收到更新通知

内容的提问来源于stack exchange，提问作者Sandeep Thomas