flask celery 使用方法

一、安装

由于celery4.0不支持window,如果在window上安装celery4.0将会出现下面的错误

flask_clery

你现在只能安装 pip install celery==3.1

二、安装py for redis 模块

pip install redis

三、安装redis服务

网上很多文章都写得模棱两可,把人坑的不要不要的!!! Redis对于Linux是官方支持的,但是不支持window,网上很多作者写文章都不写具体的系统环境,大多数直接说pip install redis就可以使用redis了,真的是坑人的玩意,本人深受其毒害

对于windows,如果你需要使用redis服务,那么进入该地址下载 https://github.com/MSOpenTech/redis/releases redis安装包,双击完成就可以了

如果你在window上不安装该redis包,将会提示 redis.exceptions.ConnectionError: Error 10061 connecting to localhost:6379. 或者 redis.exceptions.ConnectionError

需要注意是:安装目录不能安装在C盘,否则会出现权限依赖错误

四、添加redis环境变量

D:\Program Files\Redis

五、初始化redis

进入redis安装目录,打开cmd运行命令redis-server.exe redis.windows.conf,如果出错

  • 双击目录下的redis-cli.exe
  • 在出现的窗口中输入shutdown
  • 继续输入exit

六、lask 集成celyer

在Flask配置中添加配置

 # Celery 配置
CELERY_BROKER_URL = 'redis://localhost:6379/0' # broker是一个消息传输的中间件
CELERY_RESULT_BACKEND = 'redis://localhost:6379/1' # 任务执行器
  • 在flask工程的__init__目录下生产celery实例 注意!!以下代码必须在 flask app读取完配置文件后编写,否则会报错
def make_celery(app):
    celery = Celery(app.import_name, broker=app.config['CELERY_BROKER_URL'],
                    backend=app.config['CELERY_RESULT_BACKEND'])
    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


celery = make_celery(app)

完整示例如下

app = Flask(__name__)
app.config.from_object(config['default'])

def make_celery(app):
    celery = Celery(app.import_name, broker=app.config['CELERY_BROKER_URL'],
                    backend=app.config['CELERY_RESULT_BACKEND'])
    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


celery = make_celery(app)

一份比较常用的配置文件

# 注意,celery4版本后,CELERY_BROKER_URL改为BROKER_URL
BROKER_URL = 'amqp://username:passwd@host:port/虚拟主机名'
# 指定结果的接受地址
CELERY_RESULT_BACKEND = 'redis://username:passwd@host:port/db'
# 指定任务序列化方式
CELERY_TASK_SERIALIZER = 'msgpack' 
# 指定结果序列化方式
CELERY_RESULT_SERIALIZER = 'msgpack'
# 任务过期时间,celery任务执行结果的超时时间
CELERY_TASK_RESULT_EXPIRES = 60 * 20   
# 指定任务接受的序列化类型.
CELERY_ACCEPT_CONTENT = ["msgpack"]   
# 任务发送完成是否需要确认,这一项对性能有一点影响     
CELERY_ACKS_LATE = True  
# 压缩方案选择,可以是zlib, bzip2,默认是发送没有压缩的数据
CELERY_MESSAGE_COMPRESSION = 'zlib' 
# 规定完成任务的时间
CELERYD_TASK_TIME_LIMIT = 5  # 在5s内完成任务,否则执行该任务的worker将被杀死,任务移交给父进程
# celery worker的并发数,默认是服务器的内核数目,也是命令行-c参数指定的数目
CELERYD_CONCURRENCY = 4 
# celery worker 每次去rabbitmq预取任务的数量
CELERYD_PREFETCH_MULTIPLIER = 4 
# 每个worker执行了多少任务就会死掉,默认是无限的
CELERYD_MAX_TASKS_PER_CHILD = 40 
# 设置默认的队列名称,如果一个消息不符合其他的队列就会放在默认队列里面,如果什么都不设置的话,数据都会发送到默认的队列中
CELERY_DEFAULT_QUEUE = "default" 
# 设置详细的队列
CELERY_QUEUES = {
    "default": { # 这是上面指定的默认队列
        "exchange": "default",
        "exchange_type": "direct",
        "routing_key": "default"
    },
    "topicqueue": { # 这是一个topic队列 凡是topictest开头的routing key都会被放到这个队列
        "routing_key": "topic.#",
        "exchange": "topic_exchange",
        "exchange_type": "topic",
    },
    "task_eeg": { # 设置扇形交换机
        "exchange": "tasks",
        "exchange_type": "fanout",
        "binding_key": "tasks",
    },
    
}

在cmd中启动celery服务

执行命令celery -A your_application.celery worker loglevel=info,your_application为你工程的名字,在这里为 get_tieba_film

调用

@app.route('/')
@app.route('/index')
def index():
    print ("耗时的任务")
    # 任务已经交给异步处理了
    result = get_film_content.apply_async(args=[1])
    # 如果需要等待返回值,可以使用get()或wait()方法
    # result.wait()
    return '耗时的任务已经交给了celery'


@celery.task()
def get_film_content(a):
    util = SpiderRunUtil.SpiderRun(TieBaSpider.FilmSpider())
    util.start()

绑定

一个绑定任务意味着任务函数的第一个参数总是任务实例本身(self),就像 Python 绑定方法类似:

@task(bind=True) 
def add(self, x, y): 
    logger.info(self.request.id)

任务继承

任务装饰器的 base 参数可以声明任务的基类

import celery
class MyTask(celery.Task):
    def on_failure(self, exc, task_id, args, kwargs, einfo):
        print('{0!r} failed: {1!r}'.format(task_id, exc))
@task(base=MyTask)
def add(x, y):
    raise KeyError()

任务名称

每个任务必须有不同的名称。 如果没有显示提供名称,任务装饰器将会自动产生一个,产生的名称会基于这些信息: 1)任务定义所在的模块, 2)任务函数的名称

显示设置任务名称的例子:

>>> @app.task(name='sum-of-two-numbers')
>>> def add(x, y):
... return x + y
>>> add.name
'sum-of-two-numbers'

最佳实践是使用模块名称作为命名空间,这样的话如果有一个同名任务函数定义在其他模块也不会产生冲突。

>>> @app.task(name='tasks.add')
>>> def add(x, y):
... return x + y

七、安装flower

将各个任务的执行情况、各个worker的健康状态进行监控并以可视化的方式展现

pip install flower

启动flower(默认会启动一个webserver,端口为5555):

celery flower --address=127.0.0.1 --port=5555

进入http://localhost:5555即可查看。

doc

八、常见错误

ERROR/MainProcess] consumer: Cannot connect to redis://localhost:6379/0:

原因是:redis-server 没有启动 解决方案:到redis安装目录下执行redis-server.exe redis.windows.conf 检查redis是否启动:redis-cli ping

line 442, in on_task_received

解决:

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:
{'timelimit': (None, None), 'utc': True, 'chord': None, 'args': [4, 4], 'retries': 0, 'expires': None, 'task': 'main.add', 'callbacks': None,
'errbacks': None, 'taskset': None, 'kwargs': {}, 'eta': None, 'id': '97000322-93be-47e9-a082-4620e123dc5e'} (210b)
Traceback (most recent call last):
  File "d:\vm_env\flask_card\lib\site-packages\celery\worker\consumer.py", line 442, in on_task_received
    strategies[name](message, body,
KeyError: 'main.add'

原因:任务没有注册或注册不成功,只有在启动的时候提示有任务的时候,才能使用该任务

flask_celery 解决:

  1. 你在那个类中使用celery就在哪个类中执行celery -A 包名.类名.celery worker -l info
  2. 根据上一部提示的任务列表给任务设置对应的名称 如在Test中
from main import app, celery

@celery.task(name="main.Test.add".)
def add(x, y):
    print "ddddsws"
    return x + y

目录结构:

+ Card  # 工程
    + main
       + admin
          - Task.py
        - __init__.py
        - Test.py

则应该启动的命令为:

celery -A main.Test.celery worker -l info

同时,如果你的Task.py也有任务,那么你还应该重新创建一个cmd窗口执行

celery -A main.admin.Task.celery worker -l info

celery的工作进程可以创建多个

flask_celery

flask_celery

参考:

https://www.laoyuyu.me/2018/02/10/python_flask_celery/

https://www.cnblogs.com/cwp-bg/p/8759638.html

celery用户指南,强烈推荐看 redis安装 celery使用 https://redis.io/topics/quickstart

http://einverne.github.io/post/2017/05/celery-best-practice.html  Celery 最佳实践

http://orangleliu.info/2014/08/09/celery-best-practice/  Celery最佳实践-正确使用celery的7条建议

https://www.jianshu.com/p/cc3a0ffb9c76

https://windard.com/opinion/2017/03/18/Task-Queue-Celery  使用 Celery 和 redis 完成任务队列

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏存储

建立本地的Blast数据库

Blast(basic local alignment search tool) 局部序列比对基本检索工具,是NCBI开发的一款基于序列相似性的数据库搜索程序。...

59790
来自专栏求索之路

Android数据层架构的实现 上篇

最近我们app的服务器吃不消了,所以我在为服务器增加缓存层之后,又想到在app端进行二级缓存以减少app对服务器的访问。我想很多app应该在项目的初期架构的时...

34080
来自专栏步履前行

Spring Retry

  在我们的业务场景中,经常要调用其他的API来获取信息,比如我们的业务场景需要依赖个人信息来处理,这个时候调用个人信息服务的API,但是由于可能同一时段多方在...

45930
来自专栏闵开慧

tomcat6.0下找不到jasper-runtime.jar

今天有点需求,需要用jasper-runtime.jar包。但是我在我的\apache-tomcat-6.0.16\lib目录下,怎么也找不到这个jar包。结果...

36350
来自专栏Java帮帮-微信公众号-技术文章全总结

Web-第三十一天 WebService学习【悟空教程】

简单的网络应用使用单一语言写成,它的唯一外部程序就是它所依赖的数据库。大家想想是不是这样呢?

24240
来自专栏python开发教学

restful规范

          restful 规范(10) 什么是接口? - URL - 约束 # 约束继承(实现)了他的类中必须含有IFoo中的...

16930
来自专栏逸鹏说道

小解Redis 系列

官网:http://redis.io/ 推荐一个开源组件:StackExchange.Redis https://github.com/StackExchang...

31590
来自专栏bboysoul

使用colloide扫描网站后台

在渗透测试的时候我们通常要找一个网站的后台,但是有时候网站的后台并不是特别好找,比如我们学校的那个垃圾网站,今天用到的工具叫colloide,它是一款利用字典扫...

59640
来自专栏PhpZendo

深入浅出 Laravel 路由执行原理

可以说几乎所有的框架都会涉及到「路由」的处理,简单一点讲就将用户请求的 url 分配到对应的处理程序。

28930
来自专栏蓝天

Zookeeper客户端cli_st为何在crontab中运行不正常?

实践中,发现直接在命令行终端运行cli_st时,能够得到预期的结果,但一将它放到crontab中,则只收到:

11010

扫码关注云+社区

领取腾讯云代金券