My rough understanding is that Redis is better if you need the in-memory key-value store feature, however I am not sure how that has anything to do with distributing tasks?

Does that mean we should use Redis as a message broker IF we are already using it for something else?

Celery: When should you choose Redis as a message broker over RabbitMQ?

Does anyone know what the maximum value size you can store in redis? I want to use redis as a message queue with celery to store some small documents that need to be processed by a worker on another server, and I want to make sure the documents aren&#39;t going to be too big.

I found one page with a reference to 1GB, but when I followed the link on the page for where they got that answer the link wasn&#39;t valid anymore. Here is the link:  

http://news.ycombinator.com/item?id=1182005

What is the maximum value size you can store in redis?

I&#39;m trying to set up Django-Celery. I&#39;m going through the tutorial 

 http://docs.celeryproject.org/en/latest/django/first-steps-with-django.html


when I run 
    $ python manage.py celery worker --loglevel=info

I get 

    [Tasks]


    /Users/msmith/Documents/dj/venv/lib/python2.7/site-packages/djcelery/loaders.py:133:     UserWarning: Using settings.DEBUG leads to a memory leak, never use this setting in     production environments!
    warnings.warn(&#39;Using settings.DEBUG leads to a memory leak, never &#39;

    [2013-08-08 11:15:25,368: WARNING/MainProcess] /Users/msmith/Documents/dj/venv/lib/python2.7/site-packages/djcelery/loaders.py:133: UserWarning: Using settings.DEBUG leads to a memory leak, never use this setting in production environments!
    warnings.warn(&#39;Using settings.DEBUG leads to a memory leak, never &#39;

    [2013-08-08 11:15:25,369: WARNING/MainProcess] celery@sfo-mpmgr ready.
    [2013-08-08 11:15:25,382: ERROR/MainProcess] consumer: Cannot connect to     amqp://guest@127.0.0.1:5672/celeryvhost: [Errno 61] Connection refused.
    Trying again in 2.00 seconds...

has anyone encountered this issue before?

settings.py

    # Django settings for summertime project.
    import djcelery
    djcelery.setup_loader()

    BROKER_URL = &#39;amqp://guest:guest@localhost:5672/&#39;

    ...

    INSTALLED_APPS = {
        ...
        &#39;djcelery&#39;,
        &#39;celerytest&#39;
    }

wsgi.py

    import djcelery
    djcelery.setup_loader()






Django Celery - Cannot connect to amqp://guest@127.0.0.8000:5672//

I want to use [Amazon SQS][] as broker backed of [Celery][]. There’s the SQS transport implementation for [Kombu][], which Celery depends on. However there is not enough documentation for using it, so I cannot find how to configure SQS on Celery. Is there somebody that had succeeded to configure SQS on Celery?

[Amazon SQS]: http://aws.amazon.com/sqs/

[Celery]: http://celeryproject.org/
[Kombu]: https://github.com/ask/kombu

Celery with Amazon SQS

I use Celery with RabbitMQ in my Django app (on Elastic Beanstalk) to manage background tasks and I daemonized it using Supervisor.
The problem now, is that one of the period task that I defined is failing (after a week in which it worked properly), the error I&#39;ve got is:

```python
[01/Apr/2014 23:04:03] [ERROR] [celery.worker.job:272] Task clean-dead-sessions[1bfb5a0a-7914-4623-8b5b-35fc68443d2e] raised unexpected: WorkerLostError(&#39;Worker exited prematurely: signal 9 (SIGKILL).&#39;,)
Traceback (most recent call last):
  File &quot;/opt/python/run/venv/lib/python2.7/site-packages/billiard/pool.py&quot;, line 1168, in mark_as_worker_lost
    human_status(exitcode)),
WorkerLostError: Worker exited prematurely: signal 9 (SIGKILL).
```

All the processes managed by supervisor are up and running properly (`supervisorctl status` says _RUNNNING_).

I tried to read several logs on my ec2 instance but no one seems to help me in finding out what is the cause of the `SIGKILL`. What should I do? How can I investigate?

These are my **celery settings**:

```python
CELERY_TIMEZONE = &#39;UTC&#39;
CELERY_TASK_SERIALIZER = &#39;json&#39;
CELERY_ACCEPT_CONTENT = [&#39;json&#39;]
BROKER_URL = os.environ[&#39;RABBITMQ_URL&#39;]
CELERY_IGNORE_RESULT = True
CELERY_DISABLE_RATE_LIMITS = False
CELERYD_HIJACK_ROOT_LOGGER = False
```

And this is my **supervisord.conf**:

```conf
[program:celery_worker]
environment=$env_variables
directory=/opt/python/current/app
command=/opt/python/run/venv/bin/celery worker -A com.cygora -l info --pidfile=/opt/python/run/celery_worker.pid
startsecs=10
stopwaitsecs=60
stopasgroup=true
killasgroup=true
autostart=true
autorestart=true
stdout_logfile=/opt/python/log/celery_worker.stdout.log
stdout_logfile_maxbytes=5MB
stdout_logfile_backups=10
stderr_logfile=/opt/python/log/celery_worker.stderr.log
stderr_logfile_maxbytes=5MB
stderr_logfile_backups=10
numprocs=1

[program:celery_beat]
environment=$env_variables
directory=/opt/python/current/app
command=/opt/python/run/venv/bin/celery beat -A com.cygora -l info --pidfile=/opt/python/run/celery_beat.pid --schedule=/opt/python/run/celery_beat_schedule
startsecs=10
stopwaitsecs=300
stopasgroup=true
killasgroup=true
autostart=false
autorestart=true
stdout_logfile=/opt/python/log/celery_beat.stdout.log
stdout_logfile_maxbytes=5MB
stdout_logfile_backups=10
stderr_logfile=/opt/python/log/celery_beat.stderr.log
stderr_logfile_maxbytes=5MB
stderr_logfile_backups=10
numprocs=1
```

---

## Edit 1

After restarting **celery beat** the problem remains.

## Edit 2

Changed `killasgroup=true` to `killasgroup=false` and the problem remains.


Celery: WorkerLostError: Worker exited prematurely: signal 9 (SIGKILL)

I have a debian box running tasks with celery and rabbitmq for about a year. Recently I noticed tasks were not being processed so I logged into the system and noticed that celery could not connect to rabbitmq. I restarted rabbitmq-server and even though celery was not complaining anymore it was not executing new tasks now. The odd thing was that rabbitmq was devouring cpu and memory resources like crazy. Restarting server would not solve the problem. After spending couple hours looking for solution online to no avail I decided to rebuild the server.

I rebuilt new server with Debian 7.5, rabbitmq 2.8.4, celery 3.1.13 (Cipater). For about an hour or so everything worked beautifully again until celery started complaining again that it can&#39;t connect to rabbitmq!

    [2014-08-06 05:17:21,036: ERROR/MainProcess] consumer: Cannot connect to amqp://guest:**@127.0.0.1:5672//: [Errno 111] Connection refused.
    Trying again in 6.00 seconds...

I restarted rabbitmq `service rabbitmq-server start` and same issue gain:

rabbitmq started again swelling up constantly pounding on cpu and slowly taking over all ram and swap:

    PID USER      PR  NI  VIRT  RES  SHR S  %CPU %MEM    TIME+  COMMAND
    21823 rabbitmq  20   0  908m 488m 3900 S 731.2 49.4   9:44.74 beam.smp

Here&#39;s the result on `rabbitmqctl status`:

    Status of node &#39;rabbit@li370-61&#39; ...
    [{pid,21823},
     {running_applications,[{rabbit,&quot;RabbitMQ&quot;,&quot;2.8.4&quot;},
                            {os_mon,&quot;CPO  CXC 138 46&quot;,&quot;2.2.9&quot;},
                            {sasl,&quot;SASL  CXC 138 11&quot;,&quot;2.2.1&quot;},
                            {mnesia,&quot;MNESIA  CXC 138 12&quot;,&quot;4.7&quot;},
                            {stdlib,&quot;ERTS  CXC 138 10&quot;,&quot;1.18.1&quot;},
                            {kernel,&quot;ERTS  CXC 138 10&quot;,&quot;2.15.1&quot;}]},
     {os,{unix,linux}},
     {erlang_version,&quot;Erlang R15B01 (erts-5.9.1) [source] [64-bit] [smp:8:8] [async-threads:30] [kernel-poll:true]\n&quot;},
     {memory,[{total,489341272},
              {processes,462841967},
              {processes_used,462685207},
              {system,26499305},
              {atom,504409},
              {atom_used,473810},
              {binary,98752},
              {code,11874771},
              {ets,6695040}]},
     {vm_memory_high_watermark,0.3999999992280962},
     {vm_memory_limit,414559436},
     {disk_free_limit,1000000000},
     {disk_free,48346546176},
     {file_descriptors,[{total_limit,924},
                        {total_used,924},
                        {sockets_limit,829},
                        {sockets_used,3}]},
     {processes,[{limit,1048576},{used,1354}]},
     {run_queue,0},

Some entries from /var/log/rabbitmq:

    =WARNING REPORT==== 8-Aug-2014::00:11:35 ===
    Mnesia(&#39;rabbit@li370-61&#39;): ** WARNING ** Mnesia is overloaded: {dump_log,
                                                                    write_threshold}
    
    =WARNING REPORT==== 8-Aug-2014::00:11:35 ===
    Mnesia(&#39;rabbit@li370-61&#39;): ** WARNING ** Mnesia is overloaded: {dump_log,
                                                                    write_threshold}
    
    =WARNING REPORT==== 8-Aug-2014::00:11:35 ===
    Mnesia(&#39;rabbit@li370-61&#39;): ** WARNING ** Mnesia is overloaded: {dump_log,
                                                                    write_threshold}
    
    =WARNING REPORT==== 8-Aug-2014::00:11:35 ===
    Mnesia(&#39;rabbit@li370-61&#39;): ** WARNING ** Mnesia is overloaded: {dump_log,
                                                                    write_threshold}
    
    =WARNING REPORT==== 8-Aug-2014::00:11:36 ===
    Mnesia(&#39;rabbit@li370-61&#39;): ** WARNING ** Mnesia is overloaded: {dump_log,
                                                                    write_threshold}
    
    =INFO REPORT==== 8-Aug-2014::00:11:36 ===
    vm_memory_high_watermark set. Memory used:422283840 allowed:414559436
    
    =WARNING REPORT==== 8-Aug-2014::00:11:36 ===
    memory resource limit alarm set on node &#39;rabbit@li370-61&#39;.
    
    **********************************************************
    *** Publishers will be blocked until this alarm clears ***
    **********************************************************
    
    =INFO REPORT==== 8-Aug-2014::00:11:43 ===
    started TCP Listener on [::]:5672
    
    =INFO REPORT==== 8-Aug-2014::00:11:44 ===
    vm_memory_high_watermark clear. Memory used:290424384 allowed:414559436
    
    =WARNING REPORT==== 8-Aug-2014::00:11:44 ===
    memory resource limit alarm cleared on node &#39;rabbit@li370-61&#39;
    
    =INFO REPORT==== 8-Aug-2014::00:11:59 ===
    vm_memory_high_watermark set. Memory used:414584504 allowed:414559436
    
    =WARNING REPORT==== 8-Aug-2014::00:11:59 ===
    memory resource limit alarm set on node &#39;rabbit@li370-61&#39;.
    
    **********************************************************
    *** Publishers will be blocked until this alarm clears ***
    **********************************************************
    
    =INFO REPORT==== 8-Aug-2014::00:12:00 ===
    vm_memory_high_watermark clear. Memory used:411143496 allowed:414559436
    
    =WARNING REPORT==== 8-Aug-2014::00:12:00 ===
    memory resource limit alarm cleared on node &#39;rabbit@li370-61&#39;
    
    =INFO REPORT==== 8-Aug-2014::00:12:01 ===
    vm_memory_high_watermark set. Memory used:415563120 allowed:414559436
    
    =WARNING REPORT==== 8-Aug-2014::00:12:01 ===
    memory resource limit alarm set on node &#39;rabbit@li370-61&#39;.
    
    **********************************************************
    *** Publishers will be blocked until this alarm clears ***
    **********************************************************
    
    =INFO REPORT==== 8-Aug-2014::00:12:07 ===
    Server startup complete; 0 plugins started.
    
    =ERROR REPORT==== 8-Aug-2014::00:15:32 ===
    ** Generic server rabbit_disk_monitor terminating 
    ** Last message in was update
    ** When Server state == {state,&quot;/var/lib/rabbitmq/mnesia/rabbit@li370-61&quot;,
                                   50000000,46946492416,100,10000,
                                   #Ref&lt;0.0.1.79456&gt;,false}
    ** Reason for termination == 
    ** {unparseable,[]}
    
    =INFO REPORT==== 8-Aug-2014::00:15:37 ===
    Disk free limit set to 50MB
    
    =ERROR REPORT==== 8-Aug-2014::00:16:03 ===
    ** Generic server rabbit_disk_monitor terminating 
    ** Last message in was update
    ** When Server state == {state,&quot;/var/lib/rabbitmq/mnesia/rabbit@li370-61&quot;,
                                   50000000,46946426880,100,10000,
                                   #Ref&lt;0.0.1.80930&gt;,false}
    ** Reason for termination == 
    ** {unparseable,[]}
    
    =INFO REPORT==== 8-Aug-2014::00:16:05 ===
    Disk free limit set to 50MB


**UPDATE:**
Seems like problem was solved when installed newest version of rabbitmq (3.3.4-1) from rabbitmq.com repository. Originally I had one installed (2.8.4) from Debian repositories. So far rabbitmq-server is working smoothly. I will update this post if issue comes back.

**UPDATE:**
Unfortunately after about 24 hours the issue reappeared where rabbitmq shut down and restarting the process would make it consume resources until it shuts down again within minutes.  

RabbitMQ (beam.smp) and high CPU/memory load issue

If I have a function defined as follows:

    def add(x,y):
      return x+y

Is there a way to dynamically add this function as a celery PeriodicTask and kick it off at runtime? I&#39;d like to be able to do something like (pseudocode):

    some_unique_task_id = celery.beat.schedule_task(add, run_every=crontab(minute=&quot;*/30&quot;))
    celery.beat.start(some_unique_task_id)

I would also want to stop or remove that task dynamically with something like (pseudocode):

    celery.beat.remove_task(some_unique_task_id)

or

    celery.beat.stop(some_unique_task_id)

FYI I am not using djcelery, which lets you manage periodic tasks via the django admin.

How to dynamically add / remove periodic tasks to Celery (celerybeat)

It seems the longer I keep my rabbitmq server running, the more trouble I have with unacknowledged messages. I would love to requeue them. In fact there seems to be an amqp command to do this, but it only applies to the channel that your connection is using. I built a little pika script to at least try it out, but I am either missing something or it cannot be done this way (how about with rabbitmqctl?)

    import pika

    credentials = pika.PlainCredentials(&#39;***&#39;, &#39;***&#39;)
    parameters = pika.ConnectionParameters(host=&#39;localhost&#39;,port=5672,\
        credentials=credentials, virtual_host=&#39;***&#39;)

    def handle_delivery(body):
        &quot;&quot;&quot;Called when we receive a message from RabbitMQ&quot;&quot;&quot;
        print body

    def on_connected(connection):
        &quot;&quot;&quot;Called when we are fully connected to RabbitMQ&quot;&quot;&quot;
        connection.channel(on_channel_open)    

    def on_channel_open(new_channel):
        &quot;&quot;&quot;Called when our channel has opened&quot;&quot;&quot;
        global channel
        channel = new_channel
        channel.basic_recover(callback=handle_delivery,requeue=True)    
    
    try:
        connection = pika.SelectConnection(parameters=parameters,\
            on_open_callback=on_connected)    
        
        # Loop so we can communicate with RabbitMQ
        connection.ioloop.start()
    except KeyboardInterrupt:
        # Gracefully close the connection
        connection.close()
        # Loop until we&#39;re fully closed, will stop on its own
        connection.ioloop.start()

How can I recover unacknowledged AMQP messages from other channels than my connection&#39;s own?

I am trying to setup this basic example from the following doc:

&gt;http://flask.pocoo.org/docs/patterns/celery/

But so far I keep getting the below error:

&gt;AttributeError: &#39;Flask&#39; object has no attribute &#39;user_options&#39;

I am using celery 3.1.15.


    from celery import Celery

    def make_celery(app):
        celery = Celery(app.import_name, broker=app.config[&#39;CELERY_BROKER_URL&#39;])
        celery.conf.update(app.config)
        TaskBase = celery.Task
        class ContextTask(TaskBase):
            abstract = True
            def __call__(self, *args, **kwargs):
                with app.app_context():
                    return TaskBase.__call__(self, *args, **kwargs)
        celery.Task = ContextTask
        return celery

Example:
        
    from flask import Flask

    app = Flask(__name__)
    app.config.update(
        CELERY_BROKER_URL=&#39;redis://localhost:6379&#39;,
        CELERY_RESULT_BACKEND=&#39;redis://localhost:6379&#39;
    )
    celery = make_celery(app)


    @celery.task()
    def add_together(a, b):
        return a + b

Traceback error:

    Traceback (most recent call last):
      File &quot;/usr/local/bin/celery&quot;, line 11, in &lt;module&gt;
        sys.exit(main())
      File &quot;/usr/local/lib/python2.7/dist-packages/celery/__main__.py&quot;, line 30, in main
        main()
      File &quot;/usr/local/lib/python2.7/dist-packages/celery/bin/celery.py&quot;, line 81, in main
        cmd.execute_from_commandline(argv)
      File &quot;/usr/local/lib/python2.7/dist-packages/celery/bin/celery.py&quot;, line 769, in execute_from_commandline
        super(CeleryCommand, self).execute_from_commandline(argv)))
      File &quot;/usr/local/lib/python2.7/dist-packages/celery/bin/base.py&quot;, line 305, in execute_from_commandline
        argv = self.setup_app_from_commandline(argv)
      File &quot;/usr/local/lib/python2.7/dist-packages/celery/bin/base.py&quot;, line 473, in setup_app_from_commandline
        user_preload = tuple(self.app.user_options[&#39;preload&#39;] or ())
    AttributeError: &#39;Flask&#39; object has no attribute &#39;user_options&#39;

AttributeError: &#39;Flask&#39; object has no attribute &#39;user_options&#39;

What is the meaning of `bind=True` in below **celery** code? When to use it and when not?

```
@app.task(bind=True)
def send_twitter_status(self, oauth, tweet):
    try:
        twitter = Twitter(oauth)
        twitter.update_status(tweet)
    except (Twitter.FailWhaleError, Twitter.LoginError) as exc:
        raise self.retry(exc=exc)
```

What is the meaning of bind = True keyword in celery?

I am looking to use Flower (https://github.com/mher/flower) to monitor my Celery tasks in place of the django-admin as reccomended in their docs (http://docs.celeryproject.org/en/latest/userguide/monitoring.html#flower-real-time-celery-web-monitor). However, because I am new to this I am a little confused about the way Flower&#39;s page is only based on HTTP, and not HTTPS. How can I enable security for my Celery tasks such that any old user can&#39;t just visit the no-login-needed website http://flowerserver.com:5555 and change something?

I have considered Celery&#39;s [own documentation][1] on this, but they unfortunately there is no mention of how to secure Flower&#39;s api or web ui. All it says: ``[Need more text here]``

Thanks!


**Update:** My question is in part a duplicate of here: https://stackoverflow.com/questions/19300411/django-celery-flower-monitoring?rq=1#

However, I clarify his question here by asking how to run it using an environment that includes nginx, gunicorn, and celery all on the same remote machine. I too am wondering about how to set up Flower&#39;s outside accessible url, but also would prefer something like https instead of http if possible (or some way of securing the webui and accessing it remotely). I also need to know if leaving Flower running is a considerable security risk for anyone who may gain access to Flower&#39;s internal API and what the best way for securing this could be, or if it should just be disabled altogether and used just on an as-needed basis. 


  [1]: http://docs.celeryproject.org/en/latest/userguide/security.html

Celery Flower Security in Production

How can I create a wrapper that makes celery tasks look like `asyncio.Task`? Or is there a better way to integrate Celery with `asyncio`?

@asksol, the creator of Celery, [said this:](https://news.ycombinator.com/item?id=11870293):

&gt; It&#39;s quite common to use Celery as a distributed layer on top of async I/O frameworks (top tip: routing CPU-bound tasks to a prefork worker means they will not block your event loop).

But I could not find any code examples specifically for `asyncio` framework.

How to combine Celery with asyncio?

While issuing a new build to update code in workers how do I restart celery workers gracefully? 


**Edit:**
What I intend to do is to something like this.

 - Worker is running, probably uploading a 100 MB file to S3
 - A new build comes
 - Worker code has changes
 - Build script fires signal to the Worker(s)
 - Starts new workers with the new code
 - Worker(s) who got the signal after finishing the existing job exit.



How do I restart celery workers gracefully?

Is it possible to find out whether a task with a certain task id exists? When I try to get the status, I will always get pending.

    &gt;&gt;&gt; AsyncResult(&#39;...&#39;).status
    &#39;PENDING&#39;

I want to know whether a given task id is a real celery task id and not a random string. I want different results depending on whether there is a valid task for a certain id.

There may have been a valid task in the past with the same id but the results may have been deleted from the backend.

Find out whether celery task exists

I recently switch to Celery 3.0. Before that I was using [Flask-Celery][1] in order to integrate Celery with Flask. Although it had many issues like hiding some powerful Celery functionalities but it allowed me to use the full context of Flask app and especially Flask-SQLAlchemy.

In my background tasks I am processing data and the SQLAlchemy ORM to store the data. The maintainer of Flask-Celery has dropped support of the plugin. The plugin was pickling the Flask instance in the task so I could have full access to SQLAlchemy.

I am trying to replicate this behavior in my tasks.py file but with no success. Do you have any hints on how to achieve this?


  [1]: https://github.com/ask/flask-celery/

How to use Flask-SQLAlchemy in a Celery task

I am new to message brokers like [RabbitMQ](https://www.rabbitmq.com/) which we can use to create tasks / message queues for a scheduling system like [Celery](http://www.celeryproject.org/).

Now, here is the question:

- I can create a table in [PostgreSQL](https://www.postgresql.org/) which can be appended with new tasks and consumed by the consumer program like Celery.

- Why on earth would I want to setup a whole new tech for this like RabbitMQ?

Now, I believe scaling cannot be the answer since our database like PostgreSQL can work in a distributed environment.

I googled for what problems does the database poses for the particular problem, and I found:

- polling keeps the database busy and low performing
- locking of the table -&gt; again low performing
- millions of rows of tasks -&gt; again, polling is low performing

Now, how does RabbitMQ or any other message broker like that solves these problems?

Also, I found out that `AMQP` protocol is what it follows. What&#39;s great in that?

Can [Redis](https://redis.io/) also be used as a message broker? I find it more analogous to Memcached than RabbitMQ.

Please shed some light on this!

Why do we need message brokers like RabbitMQ over a database like PostgreSQL?

How can I delete all pending tasks without knowing the `task_id` for each task?

Deleting all pending tasks in celery / rabbitmq

How can I retrieve a list of tasks in a queue that are yet to be processed?

Retrieve list of tasks in a queue in Celery

The Celery documentation [mentions testing Celery within Django][1] but doesn&#39;t explain how to test a Celery task if you are not using Django. How do you do this?


  [1]: http://docs.celeryproject.org/projects/django-celery/en/2.4/cookbook/unit-testing.html

How do you unit test a Celery task?

I&#39;m trying to run [example][1] from Celery documentation.

I run: `celeryd --loglevel=INFO`

    /usr/local/lib/python2.7/dist-packages/celery/loaders/default.py:64: NotConfigured: No &#39;celeryconfig&#39; module found! Please make sure it exists and is available to Python.
      &quot;is available to Python.&quot; % (configname, )))
    [2012-03-19 04:26:34,899: WARNING/MainProcess]  
    
     -------------- celery@ubuntu v2.5.1
    ---- **** -----
    --- * ***  * -- [Configuration]
    -- * - **** ---   . broker:      amqp://guest@localhost:5672//
    - ** ----------   . loader:      celery.loaders.default.Loader
    - ** ----------   . logfile:     [stderr]@INFO
    - ** ----------   . concurrency: 4
    - ** ----------   . events:      OFF
    - *** --- * ---   . beat:        OFF
    -- ******* ----
    --- ***** ----- [Queues]
     --------------   . celery:      exchange:celery (direct) binding:celery

tasks.py:

    # -*- coding: utf-8 -*-
    from celery.task import task
    
    @task
    def add(x, y):
        return x + y

run_task.py:

    # -*- coding: utf-8 -*-
    from tasks import add
    result = add.delay(4, 4)
    print (result)
    print (result.ready())
    print (result.get())

In same folder celeryconfig.py:

    CELERY_IMPORTS = (&quot;tasks&quot;, )
    CELERY_RESULT_BACKEND = &quot;amqp&quot;
    BROKER_URL = &quot;amqp://guest:guest@localhost:5672//&quot;
    CELERY_TASK_RESULT_EXPIRES = 300

When I run &quot;run_task.py&quot;:

on python console

    eb503f77-b5fc-44e2-ac0b-91ce6ddbf153
    False

errors on celeryd server

    [2012-03-19 04:34:14,913: ERROR/MainProcess] Received unregistered task of type &#39;tasks.add&#39;.
    The message has been ignored and discarded.
    
    Did you remember to import the module containing this task?
    Or maybe you are using relative imports?
    Please see http://bit.ly/gLye1c for more information.
    
    The full contents of the message body was:
    {&#39;retries&#39;: 0, &#39;task&#39;: &#39;tasks.add&#39;, &#39;utc&#39;: False, &#39;args&#39;: (4, 4), &#39;expires&#39;: None, &#39;eta&#39;: None, &#39;kwargs&#39;: {}, &#39;id&#39;: &#39;841bc21f-8124-436b-92f1-e3b62cafdfe7&#39;}
    
    Traceback (most recent call last):
      File &quot;/usr/local/lib/python2.7/dist-packages/celery/worker/consumer.py&quot;, line 444, in receive_message
        self.strategies[name](message, body, message.ack_log_error)
    KeyError: &#39;tasks.add&#39;

Please explain what&#39;s the problem.


  [1]: http://ask.github.com/celery/getting-started/first-steps-with-celery.html#id3

Celery Received unregistered task of type (run example)

Currently I&#39;m working on python project that requires implement some background jobs (mostly for email sending and heavily database updates). I use Redis for task broker. So in this point I have two candidates: [Celery][1] and [RQ][2]. I had some experience with these job queues, but I want to ask you guys to share you experience of using this tools. So.

1. What pros and cons to use Celery vs. RQ.
2. Any examples of projects/task suitable to use Celery vs. RQ.

Celery looks pretty complicated but it&#39;s full featured solution. Actually I don&#39;t think that I need all these features. From other side RQ is very simple (e.g configuration, integration), but it seems that it lacks some useful features (e.g task revoking, code auto-reloading)

  [1]: http://celeryproject.org/
  [2]: http://python-rq.org/

Pros and cons to use Celery vs. RQ

How does one check whether a task is running in celery (specifically, I&#39;m using celery-django)?

I&#39;ve read the documentation, and I&#39;ve googled, but I can&#39;t see a call like:

    my_example_task.state() == RUNNING

My use-case is that I have an external (java) service for transcoding. When I send a document to be transcoded, I want to check if the task that runs that service is running, and if not, to (re)start it.

I&#39;m using the current stable versions - 2.4, I believe.

How to check task status in Celery?

I have been reading the doc and searching but cannot seem to find a straight answer:

Can you cancel an already executing task? (as in the task has started, takes a while, and half way through it needs to be cancelled)

I found this from the doc at [Celery FAQ][1]

    &gt;&gt;&gt; result = add.apply_async(args=[2, 2], countdown=120)
    &gt;&gt;&gt; result.revoke()

But I am unclear if this will cancel queued tasks or if it will kill a running process on a worker.  Thanks for any light you can shed!



  [1]: http://docs.celeryproject.org/en/latest/faq.html#can-i-cancel-the-execution-of-a-task

Cancel an already executing task with Celery?

I&#39;m having trouble understanding the purpose of &#39;distributed task queues&#39;. For example, python&#39;s [celery library][1]. 

I know that in celery, the python framework, you can set timed windows for functions to get executed. However, that can also be easily done in a linux crontab directed at a python script. 

And as far as I know, and shown from my own django-celery webapps, celery consumes much more RAM memory than just setting up a raw crontab. Few hundred MB difference for a relatively small app.

Can someone please help me with this distinction? Perhaps a high level explanation of how task queues / crontabs work in general would be nice also.

Thank you.


  [1]: http://www.celeryproject.org/

Distributed task queues (Ex. Celery) vs crontab scripts

From my understanding, Celery is a distributed task queue, which means the only thing that it should do is dispatching tasks/jobs to others servers and get the result back. RabbitMQ is a message queue, and nothing more. However, a worker could just listen to the MQ and execute the task when a message is received. This achieves exactly what Celery offers, so why need Celery at all?

Why use Celery instead of RabbitMQ?

I just found out about the configuration option `CELERYD_PREFETCH_MULTIPLIER` ([docs](http://docs.celeryproject.org/en/latest/configuration.html#std:setting-CELERYD_PREFETCH_MULTIPLIER)). The default is 4, but (I believe) I want the prefetching off or as low as possible. I set it to 1 now, which is close enough to what I&#39;m looking for, but there&#39;s still some things I don&#39;t understand:

 1. Why is this prefetching a good idea? I don&#39;t really see a reason for it, unless there&#39;s a lot of latency between the message queue and the workers (in my case, they are currently running on the same host and at worst might eventually run on different hosts in the same data center). The documentation only mentions the disadvantages, but fails to explain what the advantages are.

 2. Many people seem to set this to 0, expecting to be able to turn off prefetching that way (a reasonable assumption in my opinion). However, 0 means unlimited prefetching. Why would anyone ever want unlimited prefetching, doesn&#39;t that entirely eliminate the concurrency/asynchronicity you introduced a task queue for in the first place?

 3. Why can prefetching not be turned off? It might not be a good idea for performance to turn it off in most cases, but is there a technical reason for this not to be possible? Or is it just not implemented?

 4. Sometimes, this option is connected to `CELERY_ACKS_LATE`. For example. [Roger Hu writes](http://hustoknow.blogspot.com/2011/08/what-prefetch-in-celery-means.html#reserve-one-task-at-a-time) &#171;[…] often what [users] really want is to have a worker only reserve as many tasks as there are child processes. But this is not possible without enabling late acknowledgements […]&#187; I don&#39;t understand how these two options are connected and why one is not possible without the other. Another mention of the connection can be found [here](https://groups.google.com/forum/?fromgroups=#!topic/celery-users/q-PYCoKJEz4). Can someone explain why the two options are connected?

Understanding celery task prefetching

I&#39;m using celery and django-celery. I have defined a periodic task that I&#39;d like to test. Is it possible to run the periodic task from the shell manually so that I view the console output?

How can I run a celery periodic task from the shell manually?

I have a CPU intensive Celery task. I would like to use all the processing power (cores) across lots of EC2 instances to get this job done faster **(a celery parallel distributed task with multiprocessing - *I think*)**.

The terms, **threading**, **multiprocessing**, **distributed computing**, **distributed parallel processing** are all terms I&#39;m trying to understand better.

**Example task:**

      @app.task
      for item in list_of_millions_of_ids:
          id = item # do some long complicated equation here very CPU heavy!!!!!!! 
          database.objects(newid=id).save()



Using the code above ***(with an example if possible)*** how one would ago about distributed this task using Celery by allowing this one task to be split up utilising all the computing CPU power across all available machine in the cloud?


Celery parallel distributed task with multiprocessing

I am trying to come up with a testing methodology for our [django-celery][1] project.  I have read the notes in the [documentation][2], but it didn&#39;t give me a good idea of what to actually do.  I am not worried about testing the tasks in the actual daemons, just the functionality of **my** code.  Mainly I am wondering:

1. How can we bypass `task.delay()` during the test (I tried setting `CELERY_ALWAYS_EAGER = True` but it made no difference)?
2. How do we use the test settings that are recommended (if that is the best way) without actually changing our settings.py?
3. Can we still use `manage.py test` or do we have to use a custom runner?

Overall any hints or tips for testing with celery would be very helpful.


  [1]: http://celeryq.org/docs/django-celery/index.html#celery-integration-for-django
  [2]: http://celeryq.org/docs/django-celery/cookbook/unit-testing.html#unit-testing

Unit testing with django-celery?

For a task like this:

    from celery.decorators import task
    
    @task()
    def add(x, y):
        if not x or not y:
            raise Exception(&quot;test error&quot;)
        return self.wait_until_server_responds(

if it throws an exception and I want to retry it from the daemon side, how can apply an exponential back off algorithm, i.e. after `2^2, 2^3,2^4` etc seconds?   

Also is the retry maintained from the server side, such that if the worker happens to get killed then next worker that spawns will take the retry task?

Retry Celery tasks with exponential back off

In my `/etc/defaults/celeryd` config file, I&#39;ve set:


    CELERYD_NODES=&quot;agent1 agent2 agent3 agent4 agent5 agent6 agent7 agent8&quot;
    CELERYD_OPTS=&quot;--autoscale=10,3 --concurrency=5&quot;

I understand that the daemon spawns 8 celery workers, but I&#39;m fully not sure what `autoscale` and `concurrency` do together. I thought that concurrency was a way to specify the max number of threads that a worker can use and autoscale was a way for the worker to scale up and down child workers, if necessary. 

The tasks have a largish payload (some 20-50kB) and there are like 2-3 million such tasks, but each task runs in less than a second. I&#39;m seeing memory usage spike up because the broker distributes the tasks to every worker, thus replicating the payload multiple times. 

I think the issue is in the config and that the combination of workers + concurrency + autoscaling is excessive and I would like to get a better understanding of what these three options do.


Celery difference between concurrency, workers and autoscaling

How can I get the task_id value for a task from within the task? Here&#39;s my code:

    from celery.decorators import task
    from django.core.cache import cache
    
    @task
    def do_job(path):
        &quot;Performs an operation on a file&quot;
        
        # ... Code to perform the operation ...
        
        cache.set(current_task_id, operation_results)
    
The idea is that when I create a new instance of the task, I retrieve the `task_id` from the  task object. I then use the task id to determine whether the task has completed. I **don&#39;t** want to keep track of the task by the `path` value because the file is &quot;cleaned up&quot; after the task completes, and may or may not exist.

In the above example, how would I get the value of `current_task_id`?

Celery - Get task id for current task

I&#39;m running the [First Steps with Celery Tutorial][1].

We define the following task:

    from celery import Celery
    
    app = Celery(&#39;tasks&#39;, broker=&#39;amqp://guest@localhost//&#39;)
    
    @app.task
    def add(x, y):
        return x + y

Then call it:

    &gt;&gt;&gt; from tasks import add
    &gt;&gt;&gt; add.delay(4, 4)

But I get the following error:

    AttributeError: &#39;DisabledBackend&#39; object has no attribute &#39;_get_task_meta_for&#39;

I&#39;m running both the celery worker and the rabbit-mq server. Rather strangely, celery worker reports the task as succeeding:

    [2014-04-22 19:12:03,608: INFO/MainProcess] Task test_celery.add[168c7d96-e41a-41c9-80f5-50b24dcaff73] succeeded in 0.000435483998444s: 19 

Why isn&#39;t this working?

  [1]: http://docs.celeryproject.org/en/latest/getting-started/first-steps-with-celery.html


Celery with RabbitMQ: AttributeError: &#39;DisabledBackend&#39; object has no attribute &#39;_get_task_meta_for&#39;

I&#39;m trying to get Celery logging working with `Django`. I have logging set-up in `settings.py` to go to console (that works fine as I&#39;m hosting on `Heroku`). At the top of each module, I have:

    import logging
    logger = logging.getLogger(__name__)

And in my tasks.py, I have:

    from celery.utils.log import get_task_logger
    logger = get_task_logger(__name__)

That works fine for logging calls from a task and I get output like this:

    2012-11-13T18:05:38+00:00 app[worker.1]: [2012-11-13 18:05:38,527: INFO/PoolWorker-2] Syc feed is starting

But if that task then calls a method in another module, e.g. a `queryset` method, I get duplicate log entries, e.g.

    2012-11-13T18:00:51+00:00 app[worker.1]: [INFO] utils.generic_importers.ftp_processor process(): File xxx.csv already imported. Not downloaded
    2012-11-13T18:00:51+00:00 app[worker.1]: [2012-11-13 18:00:51,736: INFO/PoolWorker-6] File xxx.csv already imported. Not downloaded

I think I could use

    CELERY_HIJACK_ROOT_LOGGER = False
to just use the `Django` logging but this didn&#39;t work when I tried it and even if I did get it to work, I would lose the `&quot;PoolWorker-6&quot;` bit which I do want. (Incidentally, I can&#39;t figure out how to get the task name to display in the log entry from Celery, as [the docs][docs] seems to indicate that it should).

I suspect I&#39;m missing something simple here.

[docs]: http://docs.celeryproject.org/en/master/configuration.html?highlight=celeryd_task_log_format#std:setting-CELERYD_TASK_LOG_FORMAT

Django Celery Logging Best Practice

I was wondering what the correct approach to deploying a containerized Django app using gunicorn &amp; celery was. 

Specifically, each of these processes has a built-in way of scaling vertically, using `workers` for gunicorn and `concurrency` for celery. And then there is the Kubernetes approach to scaling using `replicas`

There is also this notion of setting workers equal to some function of the CPUs. Gunicorn recommends 
&gt;2-4 workers per core

However, I am confused what this translates to on K8s where CPU is a divisible shared resource - unless I use resoureceQuotas.

I want to understand what the Best Practice is. There are three options I can think of:

- Have single workers for gunicorn and a concurrency of 1 for celery, and scale them using the replicas? (horizontal scaling)
- Have gunicorn &amp; celery run in a single replica deployment with internal scaling (vertical scaling). This would mean setting fairly high values of workers &amp; concurrency respectively. 
- A mixed approach between 1 and 2, where we run gunicorn and celery with a small value for workers &amp; concurrency, (say 2), and then use K8s Deployment replicas to scale horizontally.

There are some questions on SO around this, but none offer an in-depth/thoughtful answer. Would appreciate if someone can share their experience.

Note: We use the default worker_class `sync` for Gunicorn

Docker/Kubernetes + Gunicorn/Celery - Multiple Workers vs Replicas?

At work, we need to build a jobs server for things like sending emails, building PDFs, crunching some data, etc.  Obviously, we&#39;d like to build on some sort of generic queueing system. I&#39;m familiar with Gearman, and that this is exact problem that it tries to solve: putting jobs on a queue where workers come to pick them up.  However, I&#39;m seeing lots of mentions of Rabbitmq and am unclear how it&#39;s used in this scenario. 

Is Rabbitmq a good framework to build a distributed jobs system on top of?  

Rabbitmq or Gearman - choosing a jobs queue

I&#39;m wondering how to setup a more specific logging system. All my tasks use

    logger = logging.getLogger(__name__)

as a module-wide logger.

I want celery to log to &quot;celeryd.log&quot; and my tasks to &quot;tasks.log&quot; but I got no idea how to get this working. Using `CELERYD_LOG_FILE` from django-celery I can route all celeryd related log messages to celeryd.log but there is no trace of the log messages created in my tasks.

Send log messages from all celery tasks to a single file

I&#39;m using [Celery][1] to manage asynchronous tasks. Occasionally, however, the celery process goes down which causes none of the tasks to get executed. I would like to be able to check the status of celery and make sure everything is working fine, and if I detect any problems display an error message to the user. From the Celery Worker documentation it looks like I might be able to use [ping][2] or [inspect][3] for this, but ping feels hacky and it&#39;s not clear exactly how inspect is meant to be used (if inspect().registered() is empty?).

Any guidance on this would be appreciated. Basically what I&#39;m looking for is a method like so:

    def celery_is_alive():
        from celery.task.control import inspect
        return bool(inspect().registered()) # is this right??


EDIT: It doesn&#39;t even look like registered() is available on celery 2.3.3 (even though the 2.1 docs list it). Maybe ping is the right answer.

EDIT: Ping also doesn&#39;t appear to do what I thought it would do, so still not sure the answer here.

  [1]: http://celeryproject.org/
  [2]: http://ask.github.com/celery/userguide/workers.html#ping
  [3]: http://ask.github.com/celery/userguide/workers.html#inspecting-workers

Detect whether Celery is Available/Running

I have a task in Celery that could potentially run for 10,000 seconds while operating normally.  However all the rest of my tasks should be done in less than one second.  How can I set a time limit for the intentionally long running task without changing the time limit on the short running tasks?

Setting Time Limit on specific task with celery

I am working on a project which scans user gmail inbox and provides a report. I have deployed it in **heroku** with following specs:

Language: **Python 2.7**

Framework: **Django 1.8**

Task scheduler: **Celery** (**Rabbitmq-bigwig** for broker url) 

Now when heroku execute it the celery is not giving me the output. On Heroku push its showing **Collectstatic configuration error**. I have tried using **whitenoise package**  

Also tried executing: **heroku run python manage.py collectstatic --dry-run --noinput**
Still getting the same error.

*$ heroku run python manage.py collectstatic --noinput* gave the following 
details of the error.

    File &quot;manage.py&quot;, line 10, in &lt;module&gt;
    execute_from_command_line(sys.argv)
    File &quot;/app/.heroku/python/lib/python2.7/site-packages/django/core/management/__init__.py&quot;, line 338, in execute_from_command_line
    utility.execute()
    File &quot;/app/.heroku/python/lib/python2.7/site-packages/django/core/management/__init__.py&quot;, line 303, in execute
    settings.INSTALLED_APPS
    File &quot;/app/.heroku/python/lib/python2.7/site-packages/django/conf/__init__.py&quot;, line 48, in __getattr__
    self._setup(name)
    File &quot;/app/.heroku/python/lib/python2.7/site-packages/django/conf/__init__.py&quot;, line 44, in _setup
    self._wrapped = Settings(settings_module)
    File &quot;/app/.heroku/python/lib/python2.7/site-packages/django/conf/__init__.py&quot;, line 92, in __init__
    mod = importlib.import_module(self.SETTINGS_MODULE)
    File &quot;/app/.heroku/python/lib/python2.7/importlib/__init__.py&quot;, line 37, in import_module
    __import__(name)
    File &quot;/app/salesblocker/__init__.py&quot;, line 5, in &lt;module&gt;
    from .celery import app as celery_app 
    File &quot;/app/salesblocker/celery.py&quot;, line 5, in &lt;module&gt;
    from celery import Celery
    File &quot;/app/.heroku/python/lib/python2.7/site-packages/celery/__init__.py&quot;, line 131, in &lt;module&gt;
    from celery import five  # noqa
    File &quot;/app/.heroku/python/lib/python2.7/site-packages/celery/five.py&quot;, line 153, in &lt;module&gt;
    from kombu.utils.compat import OrderedDict  # noqa
    File &quot;/app/.heroku/python/lib/python2.7/site-packages/kombu/utils/__init__.py&quot;, line 19, in &lt;module&gt;
    from uuid import UUID, uuid4 as _uuid4, _uuid_generate_random
    ImportError: cannot import name _uuid_generate_random


I have also tried to rollback heroku commit to previous working commit and cloned that code but on the next commit(changes:removed a media image from the media folder) its showing the same error again.

Thanks in advance



Cannot import name _uuid_generate_random in heroku django

I have RabbitMQ and Celery running locally on my Mac (OS/X 10.13.4), the following code works locally when I run add.delay(x,y):

    #!/usr/bin/env python
    from celery import Celery
    from celery.utils.log import get_task_logger

    logger = get_task_logger(__name__)

    app = Celery(&#39;tasks&#39;, \
            broker=&#39;pyamqp://appuser:xx@c2/appvhost&#39;, \
            backend=&#39;db+mysql://appuser:xx@c2/pigpen&#39;)

    @app.task(bind=True)
    def dump_context(self, x, y):
        print(&#39;Executing task id {0.id}, args: {0.args!r} kwargs {0.kwargs!r}&#39;.format(self.request))

    @app.task
    def add(x, y):
        logger.info(&#39;Adding {0} + {1}&#39;.format(x, y))
        return x + y

However when I try to run the Celery worker on an ODROID-C2 running Kali 2018.2 (w. current updates, I get the following error when running `celery -A tasks worker --loglevel=info`:

    Traceback (most recent call last):
      File &quot;/usr/local/bin/celery&quot;, line 11, in &lt;module&gt;
        sys.exit(main())
      File &quot;/usr/local/lib/python2.7/dist-packages/celery/__main__.py&quot;, line 14, in main
        _main()
      File &quot;/usr/local/lib/python2.7/dist-packages/celery/bin/celery.py&quot;, line 326, in main
        cmd.execute_from_commandline(argv)
      File &quot;/usr/local/lib/python2.7/dist-packages/celery/bin/celery.py&quot;, line 488, in execute_from_commandline
        super(CeleryCommand, self).execute_from_commandline(argv)))
      File &quot;/usr/local/lib/python2.7/dist-packages/celery/bin/base.py&quot;, line 281, in execute_from_commandline
        return self.handle_argv(self.prog_name, argv[1:])
      File &quot;/usr/local/lib/python2.7/dist-packages/celery/bin/celery.py&quot;, line 480, in handle_argv
        return self.execute(command, argv)
      File &quot;/usr/local/lib/python2.7/dist-packages/celery/bin/celery.py&quot;, line 412, in execute
        ).run_from_argv(self.prog_name, argv[1:], command=argv[0])
      File &quot;/usr/local/lib/python2.7/dist-packages/celery/bin/worker.py&quot;, line 221, in run_from_argv
        return self(*args, **options)
      File &quot;/usr/local/lib/python2.7/dist-packages/celery/bin/base.py&quot;, line 244, in __call__
        ret = self.run(*args, **kwargs)
      File &quot;/usr/local/lib/python2.7/dist-packages/celery/bin/worker.py&quot;, line 255, in run
        **kwargs)
      File &quot;/usr/local/lib/python2.7/dist-packages/celery/worker/worker.py&quot;, line 99, in __init__
        self.setup_instance(**self.prepare_args(**kwargs))
      File &quot;/usr/local/lib/python2.7/dist-packages/celery/worker/worker.py&quot;, line 122, in setup_instance
        self.should_use_eventloop() if use_eventloop is None
      File &quot;/usr/local/lib/python2.7/dist-packages/celery/worker/worker.py&quot;, line 241, in should_use_eventloop
        self._conninfo.transport.implements.async and
      File &quot;/home/autossh/.local/lib/python2.7/site-packages/kombu/transport/base.py&quot;, line 125, in __getattr__
        raise AttributeError(key)
    AttributeError: async

From the Kali ODROID I am able to connect to the RabbitMQ instance on the host named c2 using a Python Pika script and mysql from that device works to the c2 machine as well. I have found similar errors, none of those solutions have worked for me.

Celery version installed on the ODROID-C2 via pip is:

    celery --version
    4.1.0 (latentcall)


Celery AttributeError: async error

I use celery to update RSS feeds in my news aggregation site. I use one @task for each feed, and things seem to work nicely.

There&#39;s a detail that I&#39;m not sure to handle well though: all feeds are updated once every minute with a @periodic_task, but what if a feed is still updating from the last periodic task when a new one is started ? (for example if the feed is really slow, or offline and the task is held in a retry loop)

Currently I store tasks results and check their status like this:

    import socket
    from datetime import timedelta
    from celery.decorators import task, periodic_task
    from aggregator.models import Feed


    _results = {}


    @periodic_task(run_every=timedelta(minutes=1))
    def fetch_articles():
        for feed in Feed.objects.all():
            if feed.pk in _results:
                if not _results[feed.pk].ready():
                    # The task is not finished yet
                    continue
            _results[feed.pk] = update_feed.delay(feed)


    @task()
    def update_feed(feed):
        try:
            feed.fetch_articles()
        except socket.error, exc:
            update_feed.retry(args=[feed], exc=exc)

Maybe there is a more sophisticated/robust way of achieving the same result using some celery mechanism that I missed ?

Running &quot;unique&quot; tasks with celery

I am using celerybeat to kick off a primary task that kicks of a number of secondary tasks. I have both tasks written already. 

Is there a way to easily do this? Does Celery allow for tasks to be run from within tasks?

My example:

    @task
    def compute(users=None):
        if users is None:
            users = User.objects.all()
    
        tasks = []
        for user in users:
            tasks.append(compute_for_user.subtask((user.id,)))
    
        job = TaskSet(tasks)
        job.apply_async() # raises a IOError: Socket closed
        
    @task
    def compute_for_user(user_id):
        #do some stuff

`compute` gets called from celerybeat, but causes an IOError when it tries to run `apply_async`. Any ideas?

Celery task that runs more tasks

I am new to celery.I know how to install and run one server but I need to distribute the task to multiple machines.
My project uses celery to assign user requests passing to a web framework to different machines and then returns the result.
I read the documentation but there it doesn&#39;t mention how to set up multiple machines.
What am I missing?

All Celery Solutions on Gang of Coders

Total of 45 Celery Solutions