如何在Vercel的Python-Flask无服务器函数中异步发送日志到远程服务器

问题根源 #

Vercel的无服务器函数在执行完return语句后会立即回收所有资源，包括子进程、线程，这直接导致你用multiprocessing.Process启动的日志写入进程还未完成就被终止，最终出现半数日志写入失败的情况。

可行解决方案 #

方案1：同步写入+超时控制 #

如果远程日志服务响应速度尚可，可直接同步写入但设置短超时，既保证日志写入完成再返回，又避免因远程服务故障阻塞请求太久。

修改后代码：

from flask import Flask, request
from flask_cors import CORS
import requests

app = Flask(__name__)
cors = CORS(app, resources={r"/api/*": {"origins": "*"}})

def remote_log(log_data):
    try:
        # 替换为你的远程日志地址，设置2秒超时
        requests.post("https://your-log-server.com/log", json={"data": log_data}, timeout=2)
    except requests.exceptions.RequestException:
        # 写入失败时输出到Vercel内置日志
        print(f"日志写入失败: {log_data}")

@app.route('/api/chat', methods=['POST'])
def chat():
    log_data = "This is my data"
    remote_log(log_data)
    return "My response"

@app.route('/')
def home():
    return 'Server Deployed'

@app.route('/about')
def about():
    return 'About'

if __name__ == '__main__':
    app.run(debug=True)

• 优点：实现简单，能确保日志写入完成再响应
• 缺点：会增加请求响应时间（最长为超时时间）

方案2：利用Vercel内置日志系统 #

Vercel会自动收集函数的stdout/stderr输出，直接用print或logging模块输出日志即可，无需自行维护远程日志服务。

修改后代码：

from flask import Flask, request
from flask_cors import CORS
import logging

# 配置日志输出级别
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = Flask(__name__)
cors = CORS(app, resources={r"/api/*": {"origins": "*"}})

@app.route('/api/chat', methods=['POST'])
def chat():
    log_data = "This is my data"
    # 直接输出日志，Vercel自动收集持久化
    logger.info(log_data)
    return "My response"

@app.route('/')
def home():
    return 'Server Deployed'

@app.route('/about')
def about():
    return 'About'

if __name__ == '__main__':
    app.run(debug=True)

• 优点：零额外开发成本，无需维护远程服务
• 缺点：日志仅能在Vercel控制台查看，无法自定义写入私有远程服务器

方案3：异步消息队列（可靠无阻塞） #

如果必须写入私有远程服务器且不想阻塞主请求，可通过消息队列实现异步处理：主函数仅将日志消息推入队列后立即返回，由独立消费者服务负责写入远程日志。

以Serverless Redis（如Upstash）为例：

主函数代码

from flask import Flask, request
from flask_cors import CORS
import redis
import json

app = Flask(__name__)
cors = CORS(app, resources={r"/api/*": {"origins": "*"}})

# 初始化Redis客户端（替换为你的Redis地址和密码）
r = redis.Redis(host="your-redis-host", port=6379, password="your-redis-password", decode_responses=True)

@app.route('/api/chat', methods=['POST'])
def chat():
    log_data = "This is my data"
    # 将日志推入Redis队列
    r.lpush("log_queue", json.dumps({"data": log_data}))
    return "My response"

# 其他路由...

消费者代码（可部署为独立Vercel函数/定时任务）

import redis
import json
import requests

r = redis.Redis(host="your-redis-host", port=6379, password="your-redis-password", decode_responses=True)

def remote_log(log_data):
    try:
        requests.post("https://your-log-server.com/log", json={"data": log_data}, timeout=2)
    except requests.exceptions.RequestException:
        print(f"日志写入失败: {log_data}")

def consume_logs():
    while True:
        # 阻塞式取出队列消息
        message = r.brpop("log_queue", timeout=5)
        if message:
            log_data = json.loads(message[1])
            remote_log(log_data["data"])

if __name__ == '__main__':
    consume_logs()

• 优点：完全不阻塞主请求，日志写入由独立服务保证
• 缺点：需额外维护消息队列和消费者服务，复杂度较高

方案4：线程+短延迟（临时 workaround，不推荐） #

如果日志写入耗时极短，可尝试用线程替代进程，并在return前给线程少量执行时间，但此方法仍不可靠，Vercel仍可能提前终止线程：

from flask import Flask, request
from flask_cors import CORS
import time
import threading

def remote_log(log_data):
    time.sleep(1)
    print(f"日志写入完成: {log_data}")

def send_data_to_server(log_data):
    remote_log(log_data=log_data)

app = Flask(__name__)
cors = CORS(app, resources={r"/api/*": {"origins": "*"}})

@app.route('/api/chat', methods=['POST'])
def chat():
    log_data = "This is my data"
    t = threading.Thread(target=send_data_to_server, args=(log_data,), daemon=False)
    t.start()
    # 给线程0.1秒执行时间
    time.sleep(0.1)
    return "My response"

# 其他路由...

• 优点：修改成本低
• 缺点：日志仍有丢失概率，且增加响应时间

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