为什么 Nginx 会截断 gRPC 流响应？

Question

我以前问过这个问题，但决定删除那个旧问题并重新表述它连同 minimum reproducible example。问题是，当我在 nginx 上部署我的 gunicorn 网络服务器时，我通过 gRPC 来自我的 go 服务器的流式响应被截断了。所有详细信息都可以在存储库中找到。我对该站点的 nginx 配置如下所示：

server {
    listen 80 default_server;
    server_name example.com;

    location / {
    #include proxy_params;
    proxy_pass http://localhost:5000;
    proxy_buffering off;
    chunked_transfer_encoding off;
    }
}

前端接收并解析响应的代码如下所示：

        <script>
            (async function(){
                const response = await fetch("{{ url_for('realtimedata') }}");
                const reader = response.body.pipeThrough(new TextDecoderStream()).getReader();
                while (true) {
                    const {done, value} = await reader.read();
                    if (done) break;
                    try {
                        console.log('Received', value);
                        const rtd = JSON.parse(value);
                        console.log('Parsed', rtd);
                    } catch(err) {
                        console.log(err);
                    }
                }
            })()
        </script>

关于来自 go 服务器的数据需要注意的事项，一个服务提供一个包含 96 个字段的数据对象，另一个服务提供一个包含 200 个字段的数据。这使得传入的流响应具有可变长度（以字节为单位）。

我想使用gunicorn，因为我可能同时有多个监听器。使用 gunicorn 解决了一个问题，即所有响应都发送到网络服务器，但它们分布在活动客户端之间。所以每个客户都会得到不同的响应，但不是全部。

编辑： 我已经尝试将 goserver 上的响应对象大小更改为与两种服务相同，但截断仍然发生。具有可变长度似乎不是问题。我也试过用 uWSGI 而不是 gunicorn 来做这个，问题仍然存在。我什至设置了 uwsgi_buffering off;，但问题仍然存在。

更新： 我有运行使用 Apache2 而不是 Nginx 的最小可重现示例，我遇到了同样的问题。可能是其他问题。

Answer 1

查看您的 python 代码，似乎使用 websockets 可以更好地将数据从后端推送到前端。我已经重写了您的后端以使用 FastAPI 而不是 Flask 并修改了 nginx 配置。

main.py

import asyncio
import dependencies.rpc_pb2 as r
import dependencies.rpc_pb2_grpc as rpc
from fastapi import FastAPI, WebSocket, Request
from fastapi.templating import Jinja2Templates
import grpc
import json
import os

os.environ["GRPC_SSL_CIPHER_SUITES"] = 'HIGH+ECDSA'

app = FastAPI()
templates = Jinja2Templates(directory="templates")
server_addr = "localhost"
server_port = 3567

@app.get("/")
def read_root(request: Request):
    return templates.TemplateResponse("index.html", {"request": request})

def parseRtd(rtd):
    rtdDict = {}
    rtdDict["source"] = rtd.source
    rtdDict["is_scanning"] = rtd.is_scanning
    rtdDict["timestamp"] = int(rtd.timestamp)
    rtdDict["data"] = {}
    for key, v in rtd.data.items():
        rtdDict["data"][int(key)] = {"name": v.name, "value": v.value}
    return rtdDict

def get_rtd():
    channel = grpc.insecure_channel(f"{server_addr}:{server_port}")
    stub = rpc.RpcServiceStub(channel)
    for rtd in stub.SubscribeDataStream(r.SubscribeDataRequest()):
        yield parseRtd(rtd)

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    await websocket.send_json({"test": "this is a test"})
    it = get_rtd()
    while True:
        await asyncio.sleep(0.1)
        payload = next(it)
        await websocket.send_json(payload)

index.html

<html>
    <head>
        <script src="https://cdnjs.cloudflare.com/ajax/libs/socket.io/4.4.0/socket.io.js" integrity="sha512-nYuHvSAhY5lFZ4ixSViOwsEKFvlxHMU2NHts1ILuJgOS6ptUmAGt/0i5czIgMOahKZ6JN84YFDA+mCdky7dD8A==" crossorigin="anonymous" referrerpolicy="no-referrer"></script>
    </head>
    <body>
        <script>
            var ws = new WebSocket("ws://localhost:5000/ws");     
            ws.onopen = function () {
                console.log("websocket was open");
            };
            ws.onclose = () => {
                console.log("Websocket was closed!");
            }
            ws.onerror = (error) =>{
                console.error("Websocket error: " + JSON.stringify(error));
            };
            ws.onmessage = (message) => {
                console.log("MSG: " + message.data );
            };
        </script>
    </body>
</html>

webserver.conf

server {
    listen 80 default_server;
    server_name example.com;

    location / {
        include proxy_params;
        proxy_pass http://localhost:5000;
    }

    location /ws {
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header Host $host;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "Upgrade";
        proxy_pass http://localhost:5000;
    }
}

为什么 Nginx 会截断 gRPC 流响应？

Why is Nginx truncating the gRPC streaming response?

nginx

gevent

flask

gunicorn

grpc