Celery任务中执行Cassandra原生查询出现超时问题求助

你碰到的是Celery worker环境下Cassandra连接失效导致的超时问题，这在Django+Celery+Cassandra的组合里很常见——核心原因是Celery的forked worker进程复用了父进程的Cassandra会话，而fork后的会话是无法正常工作的，再加上超时参数配置位置不对，才会出现你看到的情况。下面是具体的解决步骤：

一、修复Celery Worker的Cassandra初始化逻辑 #

你当前的cassandra_init函数没有正确处理Celery worker的fork特性，导致每个worker进程拿到的是父进程遗留的无效会话。修改成以下逻辑：

from django.db import connection as django_cassandra_conn
from celery.signals import worker_process_init

# 全局变量存储集群和会话引用
cql_cluster = None
cql_session = None

def cassandra_init(*args, **kwargs):
    """为每个Celery Worker进程初始化独立的Cassandra连接"""
    global cql_cluster, cql_session

    # 安全关闭父进程遗留的旧连接（避免资源泄漏）
    if cql_session is not None:
        try:
            cql_session.shutdown()
        except Exception:
            pass
        cql_session = None
    if cql_cluster is not None:
        try:
            cql_cluster.shutdown()
        except Exception:
            pass
        cql_cluster = None

    # 强制关闭Django的旧连接，触发重新初始化
    django_cassandra_conn.close()
    # 通过获取cursor触发新连接的创建，确保当前worker进程拥有独立会话
    _ = django_cassandra_conn.cursor()

# 绑定worker进程初始化信号
worker_process_init.connect(cassandra_init)

这样做的目的是：每个Celery worker进程启动时，都会彻底清理父进程的旧连接，然后重新创建属于自己的Cassandra会话，避免fork带来的连接失效问题。

二、正确配置Cassandra超时参数 #

你之前尝试增加超时无效，是因为没在正确的位置配置。有两种方式可以设置：

1. 全局配置（所有查询生效） #

在settings.py的DATABASES配置中，通过OPTIONS添加Cassandra驱动的超时参数：

DATABASES = {
    'default': {
        'ENGINE': 'django_cassandra_engine',
        'NAME': 'your_keyspace_name',
        'HOST': '192.168.98.65',
        'OPTIONS': {
            'connection': {
                'connect_timeout': 10,  # 连接Cassandra的超时时间（秒）
                'request_timeout': 30,  # 单个查询的请求超时时间（秒）
            },
            'session': {
                'default_timeout': 30,  # 会话的默认超时时间
            }
        }
    }
}

2. 单个查询手动指定超时 #

如果只是特定查询需要调整超时，可以用SimpleStatement来指定：

from django.db import connection
from cassandra.query import SimpleStatement

cursor = connection.cursor()
# 创建带超时的查询语句（这里设置30秒超时）
query_stmt = SimpleStatement("SELECT cpu_info FROM ap_live_stats;", timeout=30)
total_ap = cursor.execute(query_stmt)

三、额外排查建议 #

• 检查Celery并发数：如果CELERY_WORKER_CONCURRENCY设置得太高，会导致Cassandra集群的连接数被耗尽，进而引发超时。可以适当降低这个值，比如设置为4或8，根据你的Cassandra集群规模调整。
• 检查Cassandra集群状态：登录到Cassandra节点，执行nodetool status确认节点是否正常在线，用nodetool tpstats查看是否有任务堆积或超时情况，同时检查节点的CPU、内存和网络是否有瓶颈。

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