अजवाइन पहला कदम - result.get पर टाइमआउट त्रुटि()

Question

मैं अजवाइन पहला कदम ट्यूटोरियल यहाँ अनुसरण कर रही हूं: http://celery.readthedocs.org/en/latest/getting-started/first-steps-with-celery.html#keeping-results

मैं ट्यूटोरियल के साथ अनुसरण कर रही हूं उसी रूप में, RabbitMQ के साथ।

जब मैं परिणाम कर रहा हूं .get (timeout = 1), यह एक टाइमआउट त्रुटि दिखा रहा है, भले ही यह एक साधारण ऐड ऑपरेशन है, और मैं देख सकता हूं कि कार्यकर्ता चल रहा है और सही परिणाम (8) का उत्पादन कर रहा है अन्य विंडो

(venv) C:\Volt\celerytest>ipython 
Python 2.7.6 (default, Nov 10 2013, 19:24:18) [MSC v.1500 32 bit (Intel)] 
Type "copyright", "credits" or "license" for more information. 

IPython 2.1.0 -- An enhanced Interactive Python. 
?   -> Introduction and overview of IPython's features. 
%quickref -> Quick reference. 
help  -> Python's own help system. 
object? -> Details about 'object', use 'object??' for extra details. 

In [1]: from tasks import add 

In [2]: a = add(1,3) 

In [3]: a 
Out[3]: 4 

In [4]: a = add.delay(1,3) 

In [5]: a.ready() 
Out[5]: False 

In [6]: a = add.delay(4,4) 

In [7]: a.get(timeout=0.5) 
--------------------------------------------------------------------------- 
TimeoutError        Traceback (most recent call last) 
<ipython-input-7-2c407a92720e> in <module>() 
----> 1 a.get(timeout=0.5) 

C:\Users\Som\Envs\venv\lib\site-packages\celery\result.pyc in get(self, timeout, 
propagate, interval, no_ack, follow_parents) 
    167     interval=interval, 
    168     on_interval=on_interval, 
--> 169     no_ack=no_ack, 
    170   ) 
    171   finally: 

C:\Users\Som\Envs\venv\lib\site-packages\celery\backends\amqp.pyc in wait_for(se 
lf, task_id, timeout, cache, propagate, no_ack, on_interval, READY_STATES, PROPA 
GATE_STATES, **kwargs) 
    155          on_interval=on_interval) 
    156    except socket.timeout: 
--> 157     raise TimeoutError('The operation timed out.') 
    158 
    159   if meta['status'] in PROPAGATE_STATES and propagate: 

TimeoutError: The operation timed out. 

In [8]: 

tasks.py फ़ाइल

from celery import Celery 

app = Celery('tasks', backend='amqp', broker='amqp://') 


@app.task 
def add(x, y): 
    return x + y 

कार्यकर्ता लॉग

[tasks] 
    . tasks.add 

[2014-07-17 13:00:33,196: INFO/MainProcess] Connected to amqp://guest:**@127.0.0 
.1:5672// 
[2014-07-17 13:00:33,211: INFO/MainProcess] mingle: searching for neighbors 
[2014-07-17 13:00:34,220: INFO/MainProcess] mingle: all alone 
[2014-07-17 13:00:34,240: WARNING/MainProcess] celery@SomsPC ready. 
[2014-07-17 13:00:34,242: INFO/MainProcess] Received task: tasks.add[85ff75d8-38 
b5-442a-a574-c8b976a33739] 
[2014-07-17 13:00:34,243: INFO/MainProcess] Task tasks.add[85ff75d8-38b5-442a-a5 
74-c8b976a33739] succeeded in 0.000999927520752s: 4 
[2014-07-17 13:00:46,582: INFO/MainProcess] Received task: tasks.add[49de7c6b-96 
72-485d-926e-a4e564ccc89a] 
[2014-07-17 13:00:46,588: INFO/MainProcess] Task tasks.add[49de7c6b-9672-485d-92 
6e-a4e564ccc89a] succeeded in 0.00600004196167s: 8 

Answer 1

मुझे 'सेलेरी फर्स्ट स्टेप्स' के माध्यम से जाने के बाद बिल्कुल वही समस्या का अनुभव होता है।

मुझे लगता है कि इसका कारण backend='amqp' सुझाया गया है।

सेटअप है कि मेरे लिए काम किया है निम्नलिखित:

app = Celery('tasks', broker='amqp://guest@localhost//') 
app.conf.CELERY_RESULT_BACKEND = 'db+sqlite:///results.sqlite' 

डॉक्स, जब AMQP परिणाम बैक-एंड प्रयोग किया जाता है के अनुसार, प्रत्येक परिणाम केवल एक बार पुनः प्राप्त किया जा सकता है (यह वास्तव में क्वेरी में एक ही संदेश है) ।

मुझे लगता है, अपने कार्यकर्ता प्रक्रिया के क्रम में इसे पुनः प्राप्त परिणाम मुद्रित करने के लिए सांत्वना देने:

Task tasks.add[49de7c6b-9672-485d-926e-a4e564ccc89a] succeeded in 0.00600004196167s: 8

ताकि आप एक ही परिणाम फिर से प्राप्त करने में विफल।

Answer 2

यदि आप this thread पर देखते हैं तो ऐसा लगता है कि --pool=solo सेटिंग भी हल करती है। यह मेरे लिए काम करता है।

Answer 3

कभी कभी मैं भी redis साथ TimeoutError प्राप्त है, इसलिए मैं सहायक समारोह लागू किया:

celery_app.update(
    redis_socket_timeout=5, 
    redis_socket_connect_timeout=5, 
) 


def run_task(task, *args, **kwargs): 
    timeout = 2 * 60 
    future = task.apply_async(args, kwargs) 
    time_end = time.time() + timeout 

    while True: 
     try: 
      return future.get(timeout=timeout) 
     except redis.TimeoutError: 
      if time.time() < time_end: 
       continue 
      raise