我正在使用
celery == 4.1.0 (latentcall)
[cqlsh 5.0.1 | Cassandra 3.11.2 | CQL spec 3.4.4 | Native protocol v4]
Python 2.7.14
我正在尝试在芹菜工作器函数中执行 Cassandra 查询。但芹菜工人收到了任务,但没有执行查询。
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"
CustomerObj = dbSession.execute(dbSession.prepare(query))
return CustomerObj
get_data.delay()
我开始使用工作:
$ celery worker -A <worker_name> -l INFO -c 1
-------------- celery@ubuntu v4.1.0 (latentcall)
---- **** -----
--- * *** * -- Linux-4.13.0-21-generic-x86_64-with-Ubuntu-17.10-artful 2018-04-20 14:31:41
-- * - **** ---
- ** ---------- [config]
- ** ---------- .> app: Woker:0x7fa4a0e6f310
- ** ---------- .> transport: redis://localhost:6379/0
- ** ---------- .> results: rpc://
- *** --- * --- .> concurrency: 1 (prefork)
-- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
--- ***** -----
-------------- [queues]
.> celery exchange=celery(direct) key=celery
[tasks]
. Worker.get_data
[2018-04-20 14:31:41,271: INFO/MainProcess] Connected to redis://localhost:6379/0
[2018-04-20 14:31:41,285: INFO/MainProcess] mingle: searching for neighbors
[2018-04-20 14:31:42,315: INFO/MainProcess] mingle: all alone
.............
[2018-04-20 14:31:42,332: INFO/MainProcess] celery@ubuntu ready.
[2018-04-20 14:31:43,823: INFO/MainProcess] Received task: <worker_name>.get_data[8de91fdf-1388-4d5c-bb22-8cb00c1c065e]
工作进程只是在那里停止。它不会执行该 SELECT 查询并提供任何数据。
有人建议我如何运行此代码来执行 Cassandra 查询。
我认为您无法全局定义dbSession
。芹菜任务可以在不同的辅助角色中运行,因此连接不能是全局的。
我可以建议两个选项:
-
在任务中创建会话。它应该有效。优点是您将为每个任务创建新会话。也许
lazy
(@LazyProperty
(应该在这里有所帮助。
您可以在工作线程 级别创建连接:尝试在工作线程启动时创建会话,可能带有
worker_init
信号 (ref(。这里的问题是,你可以concurrency
级别> 1(取决于你如何启动工作线程( - 并且你需要会话池一次提供多个芹菜任务(一次处理多个 Cassandra 会话(。
顺便说一下,你应该在python中使用global
关键字。如果您正在运行一个实例,它也可以修复。
这里有一个可能对您有所帮助的相关问题: 芹菜工人数据库连接池
祝你好运!
由于芹菜不使用应用程序的连接实例。在芹菜启动时启动新连接。下面的片段是根据卡桑德拉的芹菜文档
from celery import Celery
from celery.signals import worker_process_init, beat_init
from cassandra.cqlengine import connection
from cassandra.cqlengine.connection import (
cluster as cql_cluster, session as cql_session)
def cassandra_init(**kwargs):
""" Initialize a clean Cassandra connection. """
if cql_cluster is not None:
cql_cluster.shutdown()
if cql_session is not None:
cql_session.shutdown()
connection.setup()
# Initialize worker context for both standard and periodic tasks.
worker_process_init.connect(cassandra_init)
beat_init.connect(cassandra_init)
app = Celery()
这对我有用