多Gunicorn Worker场景下仅启动一个后台任务的实现方案咨询

方案一：独立部署Kafka消费进程（最推荐） #

把Kafka消费任务从Flask应用里拆出来，写成单独的脚本（比如kafka_consumer.py），用进程管理工具（如supervisord）同时启动Gunicorn和消费进程。两者完全解耦，互不影响，从根源上避免多Worker重复启动消费任务的问题。

示例supervisord配置：

[program:flask_app]
command=gunicorn -w 4 app:app
directory=/path/to/your/app
autostart=true
autorestart=true

[program:kafka_consumer]
command=python kafka_consumer.py
directory=/path/to/your/app
autostart=true
autorestart=true

方案二：利用Gunicorn preload + 主进程判断启动消费任务 #

Gunicorn的--preload参数会让主进程先加载应用代码，再fork出多个Worker进程。我们可以在应用初始化时，判断当前进程是否为主进程，仅在主进程中用线程启动Kafka消费任务（避免阻塞主进程）。

示例代码：

from flask import Flask
import os
import threading
from kafka import KafkaConsumer

app = Flask(__name__)

def kafka_consume_task():
    consumer = KafkaConsumer('your_topic', bootstrap_servers='kafka:9092')
    for msg in consumer:
        # 消息处理逻辑
        print(f"Received message: {msg.value.decode('utf-8')}")

# 判断是否是带preload参数启动的Gunicorn主进程
if os.environ.get('GUNICORN_CMD_ARGS') and '--preload' in os.environ.get('GUNICORN_CMD_ARGS'):
    consume_thread = threading.Thread(target=kafka_consume_task, daemon=True)
    consume_thread.start()

@app.route('/')
def hello():
    return "Hello World!"

启动命令：

gunicorn --preload -w 4 app:app

注：Gunicorn主进程仅负责管理Worker，不处理请求，因此在这里启动消费线程不会影响业务请求的处理。

方案三：自定义Gunicorn Worker，仅第一个Worker启动消费任务 #

自定义Gunicorn的Worker类，在Worker初始化阶段，只让编号为0的Worker启动Kafka消费任务，其余Worker跳过该逻辑。

示例代码：

from flask import Flask
from gunicorn.workers.sync import SyncWorker
import threading
from kafka import KafkaConsumer

app = Flask(__name__)

def kafka_consume_task():
    consumer = KafkaConsumer('your_topic', bootstrap_servers='kafka:9092')
    for msg in consumer:
        # 消息处理逻辑
        print(f"Received message: {msg.value.decode('utf-8')}")

class CustomWorker(SyncWorker):
    def init_process(self):
        # 仅编号为0的Worker启动消费任务
        if self.nr == 0:
            consume_thread = threading.Thread(target=kafka_consume_task, daemon=True)
            consume_thread.start()
        super().init_process()

@app.route('/')
def hello():
    return "Hello World!"

启动命令：

gunicorn -w 4 -k app:CustomWorker app:app

注：如果Worker异常重启，新启动的Worker若编号变为0会重新启动消费任务，需做好消息处理的幂等性。

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