React+Nest.js WebSocket首条消息接收/实时推送失效问题求助

我搭建了基于React.js客户端与Nest.js服务端的应用，目标是通过WebSocket实现：客户端向服务端发送数据，服务端执行复杂计算，在每次迭代后向客户端推送计算进度，最终发送结果。

最初遇到循环推送时客户端仅收第一条消息，简化代码后该问题解决，但加入每次耗时约5秒的复杂计算后，客户端无法实时接收消息，而是在循环结束后一次性收到所有消息。请问如何实现实时推送？

相关代码 #

客户端代码 #

const socket = io('http://localhost:5000');

socket.emit('correlate', {
    formData
});

socket.on('correlate', (d) => {
   console.log(d); // 这里只打印第一条消息
});

服务端简化版代码 #

import {
    ConnectedSocket,
    MessageBody,
    SubscribeMessage,
    WebSocketGateway,
    WebSocketServer,
} from '@nestjs/websockets';
import * as papa from 'papaparse';
import { Server } from 'socket.io';
import {AppService} from "./app.service";
import {HttpException} from "@nestjs/common";

@WebSocketGateway({
    cors: {
        origin: '*',
    },
    maxHttpBufferSize: 500 * 1024 * 1024
})
export class EventsGateway {
    constructor(
        private readonly appService: AppService
    ) {
    }

    @WebSocketServer()
    server: Server;

    @SubscribeMessage('correlate')
    async correlate(@MessageBody() data: any, @ConnectedSocket() client) {
        for(let i=0; i<10; i++) {
            client.emit('correlate', 'test' + i);
        }
    }
}

含复杂计算的服务端代码 #

@SubscribeMessage('correlate')
async correlate(@MessageBody() data: any, @ConnectedSocket() client) {
    for(let i=0; i<10; i++) {
        // 一些复杂、耗时的计算，每次约5秒
        client.emit('correlate', 'test' + i);
    }
}

问题出在Node.js的事件循环被同步计算阻塞了：你的复杂计算是CPU密集型的同步任务，一旦开始跑就会占满整个线程，导致WebSocket的推送消息根本没机会被发送出去，只能等整个循环跑完、事件循环空闲了，才会把所有消息一次性发出去。

要实现实时推送，得让事件循环有间隙处理WebSocket的发送操作，下面是几种可行的方法：

方法1：把复杂计算改成异步，给事件循环腾时间 #

把每次计算用Promise包起来，用await等待计算完成，这样事件循环就能在计算间隙处理推送了：

@SubscribeMessage('correlate')
async correlate(@MessageBody() data: any, @ConnectedSocket() client) {
    for(let i=0; i<10; i++) {
        // 用Promise包裹复杂计算，让事件循环能处理其他任务
        await new Promise(resolve => {
            // 这里放你的复杂计算逻辑
            // some complex, long computations, about 5s per iteration
            
            resolve();
        });
        
        // 计算完立刻推进度
        client.emit('correlate', 'test' + i);
    }
}

如果你的计算本身就是异步函数（比如返回Promise），直接await调用就行，不用额外包Promise。

方法2：用setImmediate让出事件循环 #

要是不想改计算逻辑的结构，可以用递归+setImmediate，让每次迭代之间给事件循环留处理消息的时间：

@SubscribeMessage('correlate')
async correlate(@MessageBody() data: any, @ConnectedSocket() client) {
    const runIteration = async (i) => {
        // 先做复杂计算
        // some complex, long computations, about 5s per iteration
        
        // 推进度
        client.emit('correlate', 'test' + i);
        
        // 还没到最后一次的话，等事件循环空闲了再跑下一次
        if(i < 9) {
            setImmediate(() => runIteration(i+1));
        }
    };
    
    await runIteration(0);
}

方法3：用任务队列彻底隔离计算（推荐） #

如果计算真的很耗时，最好把计算任务放到单独的队列里，别让它阻塞WebSocket网关。比如用Nest.js的@nestjs/bull：

1. 安装并配置Bull队列
2. 把复杂计算逻辑写到队列处理器里
3. 网关收到请求后，把任务丢进队列，处理器计算完就通过WebSocket推进度

这样计算任务和WebSocket服务完全隔离，网关不会被阻塞，推送自然就实时了。

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