celery rabbit mq 详解

from celery import Celery
app = Celery('tasks',broker='redis://:alex3714@192.168.14.52')   #‘tasks’给app起的名字
backend='redis://alex3714@localhost')#将结果写入到哪

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

@app.task
def cmd(cmd_str):
    print('running cmd',cmd_str)

执行这个脚本: celery -A celery_test -l debug #当执行完这条命令后就会进入监听状态

#celery_test不用写后缀名， 指定模式。

调度端执行,会形成异步执行

from celery_test import add,cmd
add(4,5)  #这么执行的话只是本地显示结果。
add.delay(45,2) #让worker端执行
t1 = add.delay(45,2)当你赋值的时候worker端还会在执行一次
t1.get() #获取执行的结果

#这次我们测试这个任务是怎么分发的。
t2 = add.delay(22,44)
t2 = add.delay(22,44)
t2 = add.delay(22,44)
t2 = add.delay(22,44)
#当你执行的很快的时候你会发现这个任务和rabbit MQ一样是抢任务的。


from celery import Celery
app = Celery('tasks',broker='redis://:alex3714@192.168.14.52')   #‘tasks’给app起的名字
backend='redis://alex3714@localhost')#将结果写入到哪

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

@app.task
def cmd(cmd_str):
    print('running cmd',cmd_str)
    time.sleep(10)
    return time.time()

from celery_test import add,cmd
t1 = cmd.delay('dsfsdf')
t1.get()  #这时你会发现程序是阻塞的，并没有实现异步的效果
#当然这个解决也是非常简单的
t1.get(timeout=1,propa)   #这时你会发现提示你任务没有完成
t1.ready()      #可以查看任务没有没有完成，返回布尔值的状态

from celery import Celery 
app = Celery('proj',include=['proj.tasks'])  #app是Celery类的实例，创建的时候添加了proj.tasks这个模块，也就是包含了proj/tasks.py这个文件。
app.config_from_object('proj.celeryconfig')把Celery配置存放进proj/celeryconfig.py文件，使用app.config_from_object加载配置。
if __name__ == '__main__'
    app.start()

from __future__ import absolute_import
from proj.celery import app

@app.task  #让这个任务生效的装饰器
def add(x, y):
    return x + y

BROKER_URL = 'amqp://dongwm:123456@localhost:5672/web_develop' # 使用RabbitMQ作为消息代理

CELERY_RESULT_BACKEND = 'redis://localhost:6379/0' # 把任务结果存在了Redis

CELERY_TASK_SERIALIZER = 'msgpack' # 任务序列化和反序列化使用msgpack方案

CELERY_RESULT_SERIALIZER = 'json' # 读取任务结果一般性能要求不高，所以使用了可读性更好的JSON

CELERY_TASK_RESULT_EXPIRES = 60 * 60 * 24 # 任务过期时间，不建议直接写86400，应该让这样的magic数字表述更明显

CELERY_ACCEPT_CONTENT = ['json', 'msgpack'] # 指定接受的内容类型

from celery_test import add
r = add.delay(1,3) #在任务端执行一次
r #执行第二次
r.result #返回结果
r.status #返回状态
r.successful()  #去执行结果
r.backend  #保存到redis中

通过IPython触发的任务就完成了。任务的结果都需要根据上面提到的task_id获得，我们还可以用如下两种方式随时找到这个结果：

task_id = '93288a00-94ee-4727-b815-53dc3474cf3f'

In : add.AsyncResult(task_id).get()

Out: 4

或者：

In : from celery.result import AsyncResult

In : AsyncResult(task_id).get()

Out: 4