启动Django项目时并行运行Kafka监听程序的问题排查

核心问题分析 #

你遇到的问题本质是阻塞式代码的执行顺序冲突：Django的runserver和Kafka监听循环都是无限阻塞的操作，如果直接串行调用，后执行的代码永远得不到启动机会。另外使用multiprocessing时，Windows系统下未遵守if __name__ == '__main__':的强制规则，会导致进程重复启动、异常退出。

正确实现方式 #

方案1：用threading启动Kafka监听（轻量场景首选） #

修改manage.py，先启动后台线程运行Kafka监听，再启动Django服务：

import os
import sys
import threading
from run_kafka import start_kafka_listener  # 导入你的Kafka监听函数

def main():
    # 启动Kafka守护线程，Django退出时自动终止
    kafka_thread = threading.Thread(target=start_kafka_listener, daemon=True)
    kafka_thread.start()

    # 启动Django服务
    os.environ.setdefault('DJANGO_SETTINGS_MODULE', '你的项目名.settings')
    try:
        from django.core.management import execute_from_command_line
    except ImportError as exc:
        raise ImportError(
            "无法导入Django，请确认已安装并配置环境变量，或激活虚拟环境"
        ) from exc
    execute_from_command_line(sys.argv)

if __name__ == '__main__':
    main()

run_kafka.py中的监听函数需保持无限循环：

from confluent_kafka import Consumer, KafkaError

def start_kafka_listener():
    conf = {
        'bootstrap.servers': 'localhost:9092',
        'group.id': 'django-kafka-group',
        'auto.offset.reset': 'earliest'
    }

    consumer = Consumer(conf)
    consumer.subscribe(['你的主题名'])

    while True:
        msg = consumer.poll(1.0)
        if msg is None:
            continue
        if msg.error():
            if msg.error().code() == KafkaError._PARTITION_EOF:
                continue
            print(f"Kafka错误: {msg.error()}")
            break
        print(f"收到消息: {msg.value().decode('utf-8')}")
    consumer.close()

方案2：用multiprocessing启动Kafka监听（高资源占用场景） #

多进程需严格遵守Windows系统的进程启动规则，修改manage.py如下：

import os
import sys
import multiprocessing
from run_kafka import start_kafka_listener

def main():
    # 启动Kafka子进程
    kafka_process = multiprocessing.Process(target=start_kafka_listener)
    kafka_process.start()

    # 启动Django服务
    os.environ.setdefault('DJANGO_SETTINGS_MODULE', '你的项目名.settings')
    try:
        from django.core.management import execute_from_command_line
    except ImportError as exc:
        raise ImportError(
            "无法导入Django，请确认已安装并配置环境变量，或激活虚拟环境"
        ) from exc
    execute_from_command_line(sys.argv)

    # Django退出时终止Kafka进程
    kafka_process.terminate()
    kafka_process.join()

if __name__ == '__main__':
    # Windows下必须添加此行，避免多进程启动时重复执行代码
    multiprocessing.freeze_support()
    main()

关键注意事项 #

1. 禁止在Django请求处理逻辑中启动Kafka监听，必须在manage.py的启动入口提前启动。
2. 生产环境不建议用这种绑定启动的方式，应将Django和Kafka监听服务分开部署，用supervisor或systemd分别管理两个进程，避免互相影响。
3. 确保Kafka监听函数没有提前退出的逻辑，循环必须是持续运行的。

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