在Celery任务中的每个查询上使用Python cassandra驱动程序OperationTimeOut [英] Python cassandra-driver OperationTimeOut on every query in Celery task

问题描述

我在芹菜任务中异步执行的每个插入查询(小查询)都有问题.在同步模式下，当我确实插入所有内容时，效果很好，但是当它在apply_async()中执行时，我得到了:

I have a problem with every insert query (little query) which is executed in celery tasks asynchronously. In sync mode when i do insert all done great, but when it executed in apply_async() i get this:

OperationTimedOut('errors=errors=errors={}, last_host=***.***.*.***, last_host=None, last_host=None',)

跟踪:

Traceback (most recent call last):
  File "/var/nfs_www/***/env_v0/local/lib/python2.7/site-packages/celery/app/trace.py", line 240, in trace_task
    R = retval = fun(*args, **kwargs)
  File "/var/nfs_www/***/env_v0/local/lib/python2.7/site-packages/celery/app/trace.py", line 437, in __protected_call__
    return self.run(*args, **kwargs)
  File "/var/nfs_www/***/www_v1/app/mods/news_feed/tasks.py", line 26, in send_new_comment_reply_notifications
    send_new_comment_reply_notifications_method(comment_id)
  File "/var/nfs_www/***www_v1/app/mods/news_feed/methods.py", line 83, in send_new_comment_reply_notifications
    comment_type='comment_reply'
  File "/var/nfs_www/***/www_v1/app/mods/news_feed/models/storage.py", line 129, in add
    CommentsFeed(**kwargs).save()
  File "/var/nfs_www/***/env_v0/local/lib/python2.7/site-packages/cqlengine/models.py", line 531, in save
    consistency=self.__consistency__).save()
  File "/var/nfs_www/***/env_v0/local/lib/python2.7/site-packages/cqlengine/query.py", line 907, in save
    self._execute(insert)
  File "/var/nfs_www/***/env_v0/local/lib/python2.7/site-packages/cqlengine/query.py", line 786, in _execute
    tmp = execute(q, consistency_level=self._consistency)
  File "/var/nfs_www/***/env_v0/local/lib/python2.7/site-packages/cqlengine/connection.py", line 95, in execute
    result = session.execute(query, params)
  File "/var/nfs_www/***/env_v0/local/lib/python2.7/site-packages/cassandra/cluster.py", line 1103, in execute
    result = future.result(timeout)
  File "/var/nfs_www/***/env_v0/local/lib/python2.7/site-packages/cassandra/cluster.py", line 2475, in result
    raise OperationTimedOut(errors=self._errors, last_host=self._current_host)
OperationTimedOut: errors={}, last_host=***.***.*.***

有人对问题有想法吗?

我发现了这个何时cassandra-driver正在执行查询，cassandra-driver返回错误OperationTimedOut ，但是我的查询非常少，并且仅在芹菜任务中出现问题.

I found this When cassandra-driver was executing the query, cassandra-driver returned error OperationTimedOut, but my query is very little and problem only in celery tasks.

更新:

我做了一个测试任务，它也引发了这个错误.

I made a test task and it raises this error too.

@celery.task()
def test_task_with_cassandra():
    from app import cassandra_session
    cassandra_session.execute('use news_feed')
    return 'Done'

更新2:做这个:

@celery.task()
def test_task_with_cassandra():
    from cqlengine import connection
    connection.setup(app.config['CASSANDRA_SERVERS'], port=app.config['CASSANDRA_PORT'],
                     default_keyspace='test_keyspace')
    from .models import Feed
    Feed.objects.count()
    return 'Done'

知道了:

NoHostAvailable('Unable to connect to any servers', {'***.***.*.***': OperationTimedOut('errors=errors=Timed out creating connection, last_host=None, last_host=None',)})

从外壳我可以连接到它

更新3 :从github问题上的已删除线程中(在我的电子邮件中找到了此内容):(这也对我有用)本质上，这是将CQLengine插入Celery的方法:

UPDATE 3: From deleted thread on github issue (found this in my emails): (this worked for me too) Here's how, in substance, I plug CQLengine to Celery:

from celery import Celery
from celery.signals import worker_process_init, beat_init
from cqlengine import connection
from cqlengine.connection import (
    cluster as cql_cluster, session as cql_session)

def cassandra_init():
    """ 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()

这很粗糙，但是行得通.我们应该在FAQ中添加此代码段吗?

This is crude, but works. Should we add this snippet in the FAQ ?

推荐答案

我遇到了类似的问题.它似乎与在任务之间共享Cassandra会话有关.我通过为每个线程创建一个会话来解决它.确保从任务中调用 get_session()，然后执行以下操作:

I had a similar issue. It seemed to be related to sharing the Cassandra session between tasks. I solved it by creating a session per thread. Make sure you call get_session() from you tasks and then do this:

thread_local = threading.local()

def get_session():
    if hasattr(thread_local, "cassandra_session"):
        return thread_local.cassandra_session

    cluster = Cluster(settings.CASSANDRA_HOSTS)
    session = cluster.connect(settings.CASSANDRA_KEYSPACE)

    thread_local.cassandra_session = session

    return session

这篇关于在Celery任务中的每个查询上使用Python cassandra驱动程序OperationTimeOut的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！