Django用Celery异步流式写用户日志到BigQuery是否合理

你的异步解耦核心思路完全可行，Celery承载每日1万次任务属于极低负载场景，不存在选型问题，但你当前逐行写入的任务代码有可优化空间，调整后能进一步降低资源消耗、提升运行稳定性。

Celery负载能力说明 #

• 1万次/天的任务量对Celery来说完全是入门级负载：按单条任务最长耗时1.5秒计算，单Worker单并发一天理论就能处理5.7万次任务，就算所有请求都堆在1小时的峰值时段打过来，开2个Worker进程就能完全消化，根本不会出现负载打满的情况。只要用常规的Redis/RabbitMQ做消息中间件，不和核心业务抢资源，这套架构跑很长时间都不会有性能问题。

当前实现的问题 #

你现在逐行调用BigQuery流写入接口的写法虽然能跑，但存在两个明显问题：

• 写入效率偏低：BigQuery流写入接口单次请求最多支持提交1万行数据，批量提交1000行的总耗时和提交1行的耗时基本一致，都在0.5-1.5秒区间，逐行写等于平白浪费了几千次API调用额度，还更容易触发BigQuery的单表写入频率配额限制。
• 示例代码存在错误：任务装饰器@celery_app.task后面应该跟括号不是冒号；另外不建议全局初始化BigQuery客户端给所有任务复用，最好在任务函数内部初始化客户端，避免多Worker进程下的连接异常。

修正后的基础任务写法参考：

@celery_app.task(bind=True, max_retries=3, default_retry_delay=5)
def stream_bq_row(self, table_id, log):
    from google.cloud import bigquery
    client = bigquery.Client()
    try:
        client.insert_rows_json(table_id, [log])
    except Exception as e:
        # 遇到限流、网络波动时自动重试
        self.retry(exc=e)

可选优化方向 #

可以根据自己的运维成本接受度选方案：

• 最小改动方案：就用单条日志触发任务的逻辑，配上上面写的重试机制，再加个死信队列存重试多次仍然失败的日志，后续手动补数就行，不用改核心逻辑就能稳定运行，对业务接口的性能影响基本为0。
• 效率最优方案：改成攒批写入，不用每次来日志就立刻调BigQuery接口。可以在Worker侧维护一个内存缓冲队列，攒够500条日志、或者距离上次写入超过15秒就批量提交一次，这样一天下来只需要调用几十次BigQuery接口，写入延迟最高也就十几秒，完全满足近实时查日志的需求，资源消耗能降99%以上。

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