Celery Worker接收任务但未执行Cassandra查询问题求助

你的环境信息 #

• Celery == 4.1.0 (latentcall)
• Cassandra 3.11.2（cqlsh 5.0.1，CQL spec 3.4.4，Native protocol v4）
• Python 2.7.14

问题代码片段（tasks.py） #

from cassandra.cluster import Cluster
from celery import Celery

app = Celery('<workername>', backend="rpc://", broker='redis://localhost:6379/0')
dbSession = Cluster().connect()

@app.tasks()
def get_data():
    query = "SELECT * FROM customers"
    # 后续查询执行逻辑...

几个关键问题点和解决办法 #

1. 任务装饰器写错了！ #

你这里用的@app.tasks()是错误的，Celery的任务装饰器是单数的@app.task()（没有末尾的s）。这个小失误会导致Celery无法识别这个函数是可执行的任务，所以Worker虽然能接收到任务，但根本不会去执行它。赶紧把装饰器改过来：

@app.task()  # 去掉s，改成单数
def get_data():
    query = "SELECT * FROM customers"
    # 后续逻辑

2. Cassandra会话初始化的时机不对 #

你在模块加载阶段就全局初始化了dbSession = Cluster().connect()，但Celery Worker是多进程/多线程架构，全局的数据库连接在子进程里很容易失效、出现连接异常，甚至导致任务静默失败。

正确的做法是在任务内部初始化数据库会话，确保每个任务执行时都有可用的连接：

@app.task()
def get_data():
    # 任务内部创建Cassandra连接
    cluster = Cluster()
    session = cluster.connect('your_keyspace_name')  # 记得指定你的Keyspace
    query = "SELECT * FROM customers"
    result = session.execute(query)
    # 处理查询结果...
    
    # 任务结束后关闭连接
    cluster.shutdown()

如果想复用连接，可以自定义Celery的Task类，在Worker进程启动时初始化连接，这样每个进程有独立的连接实例，避免多进程共享连接的问题。

3. 忘记指定Cassandra Keyspace #

你的查询语句SELECT * FROM customers没有指定Keyspace，Cassandra要求必须明确指定要操作的Keyspace才能执行表查询。解决方法二选一：

• 连接时指定Keyspace：session = cluster.connect('your_keyspace_name')
• 查询语句中带上Keyspace前缀：query = "SELECT * FROM your_keyspace_name.customers"

如果没指定Keyspace，执行查询时会抛出异常，要是你没开启Worker日志，就看不到错误，误以为任务没执行。

4. 开启Worker日志排查细节 #

强烈建议启动Worker时开启详细日志，这样能看到任务执行时的所有错误信息，比如连接失败、权限问题、语法错误等：

celery -A tasks worker --loglevel=info

通过日志你能快速定位到任务到底是哪里出了问题，比瞎猜高效多了。

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