zoukankan      html  css  js  c++  java
  • python 关于celery的异步任务队列的基本使用(celery+redis)【无配置文件设置】

    环境说明:

    window7 X64

    python 2.7.6 、celery 3.1.25、redis 2.10.6

    本地安装的redis服务端版本号:Redis-x64-3.2.100

    工程结构说明:源文件下载请访问https://i.cnblogs.com/Files.aspx

    1、tasks.py:实例化celery,并定义生成任务的方法add()

    # -*- coding: utf-8 -*-
    
    '''
    Created on 2019年8月27日
    
    @author: lenovo
    '''
    import  time
    from celery import Celery
    from celery.bin.multi import celery_exe
    
    #实例化一个celery
    broker='redis://localhost:6379/1' #接收发送过来的任务,并等待celery的worker进行消费
    backend='redis://localhost:6379/2' #celery的worker消费完任务后,backend保存任务执行结果
    app=Celery('my_task',broker=broker,backend=backend)
    
    #在add()方法上方添加装饰器,将该方法转换为异步的
    @app.task 
    def add(x,y):
        print 'enter call func...'
        time.sleep(4)
        return x+y

     2、app.py:

    # -*- coding: utf-8 -*-
    
    '''
    Created on 2019年8月27日
    
    @author: lenovo
    '''
    from tasks import add
    
    if __name__ == '__main__':
        print 'start task...'
        result=add.delay(3, 8) #将任务发送给tasks.py中celery的broker
        print 'end task...'
        print result

    3、执行app.py生成新任务,在未启动worker前,该任务保存在broker指定的redis数据库中等待worker进行消费;

    {"body": "gAJ9cQEoVQdleHBpcmVzcQJOVQN1dGNxA4hVBGFyZ3NxBEsDSwiGcQVVBWNob3JkcQZOVQljYWxsYmFja3NxB05VCGVycmJhY2tzcQhOVQd0YXNrc2V0cQlOVQJpZHEKVSQyMTlhOWU2My0yZjExLTRiMmEtYmNjZi0yYTY2MTMzZjY3NGRxC1UHcmV0cmllc3EMSwBVBHRhc2txDVUJdGFza3MuYWRkcQ5VCXRpbWVsaW1pdHEPTk6GVQNldGFxEE5VBmt3YXJnc3ERfXESdS4=", "headers": {}, "content-type": "application/x-python-serialize", "properties": {"body_encoding": "base64", "correlation_id": "219a9e63-2f11-4b2a-bccf-2a66133f674d", "reply_to": "fa6b65cc-0efa-3ddc-bd74-246023f65e8d", "delivery_info": {"priority": 0, "routing_key": "celery", "exchange": "celery"}, "delivery_mode": 2, "delivery_tag": "aa0f8035-a256-46b3-84a9-acf8e9d490db"}, "content-encoding": "binary"}

     4、启动worker,消费该任务;

    cmd到celery实例化文件tasks.py所在路径下,执行命令:python -m celery -A tasks worker --loglevel=info 或者 celery -A tasks worker --loglevel=info

    从日志打印和下图redis数据库可以看到,worker启动后,立刻从redis中将未消费的任务进行了消费;

     任务执行的结果保存在backend设定的redis数据库中,如下图所示:

    5、测试新任务:

    发送任务:

    查看worker执行情况:

    查看backend保存结果:

    6、异步与同步效果比对:

    同步处理任务:

    工程结构:

    app.py:同步处理任务

    # -*- coding: utf-8 -*-
    import time
    
    '''
    Created on 2019年8月27日
    
    @author: lenovo
    '''
    def add(x,y):
        print 'enter call func...'
        time.sleep(4)
        return x+y
    
    
    if __name__ == '__main__':
        print 'start task...'
        result=add(2, 8)
        print 'end task...'
        print result
        

    异步处理任务:

    可以看出,同步执行任务时需要等待任务过程执行完毕后才会继续,执行任务过程中存在阻塞现象;而采用异步处理,生成任务后里面就结束了,执行任务不存在阻塞现象;

  • 相关阅读:
    kubespray -- k8s集群dashboard 访问方式
    calico 原理分析
    云主机被拿去挖矿,cpu暴涨,tcp连接突增
    监控虚拟机跟外部的tcp连接
    rook 排错记录 + Orphaned pod found kube-controller-manager的日志输出
    修改windows7本地策略--不能挂载磁盘和复制 -- 黏贴板-驱动器映射
    MySQL数据库锁机制之MyISAM引擎表锁和InnoDB行锁详解
    curl配置host
    博客收藏
    502的几种解决方案
  • 原文地址:https://www.cnblogs.com/apple2016/p/11419464.html
Copyright © 2011-2022 走看看