【Python】轻量级分布式任务调度系统-RQ

本文涉及的产品
云数据库 Tair(兼容Redis),内存型 2GB
Redis 开源版,标准版 2GB
推荐场景:
搭建游戏排行榜
简介: 一 前言       Redis Queue 一款轻量级的P分布式异步任务队列,基于Redis作为broker,将任务存到redis里面,然后在后台执行指定的Job。就目前而言有三套成熟的工具celery,huey ,rq 。
一 前言   
   Redis Queue 一款轻量级的P分布式异步任务队列,基于Redis作为broker,将任务存到redis里面,然后在后台执行指定的Job。就目前而言有三套成熟的工具celery,huey ,rq 。按照功能和使用复杂度来排序的话也是 celery>huey>rq. 因为rq 简单,容易上手,所以自己做的系统也会使用RQ作为分布式任务调度系统。
二 安装 
   因为RQ 依赖于Redis 故需要安装版本>= 2.6.0.具体安装方法请参考《Redis初探》。*nix 系统环境下安装RQ:
  1. pip install rq
无需其他配置即可以使用RQ。
三 原理
   RQ 主要由三部分构成 Job ,Queues,Worker 构成。job也就是开发定义的函数用来实现具体的功能。调用RQ 把job 放入队列Queues,Worker 负责从redis里面获取任务并执行,根据具体情况返回函数的结果。
3.1  关于job
   一个任务(job)就是一个Python对象,具体表现为在一个工作(后台)进程中异步调用一个函数。任何Python函数都可以异步调用,简单的将函数与参数追加到队列中,这叫做入队(enqueueing)。
3.2 关于Queue
   将任务加入到队列之前需要初始化一个连接到指定Redis的Queue

  1. q=Queue(connection=redis_conn)
  2. from rq_test import hello
  3. result = q.enqueue(hello,'yangyi')
   queue有如下属性:
   timeout :指定任务最长执行时间,超过该值则被认为job丢失,对于备份任务 需要设置一个比较长的时间 比如24h。
   result_ttl :存储任务返回值的有效时间,超过该值则失效。
   ttl :specifies the maximum queued time of the job before it'll be cancelled
   depends_on :specifies another job (or job id) that must complete before this job will be queued
   job_id : allows you to manually specify this job's job_id
   at_front :will place the job at the front of the queue, instead of the back
   kwargs and args : lets you bypass the auto-pop of these arguments, ie: specify a timeout argument for the underlying job function.
  需要关注的是 depends_on ,通过该属性可以做级联任务A-->B ,只有当A 执行成功之后才能执行B .
  通过指定队列的名字,我们可以把任务加到一个指定的队列中:
  1. q = Queue("low", connection = redis_conn)
  2. q.enqueue(hello, "杨一"
 对于例子中的Queue("low"),具体使用的时候可以替换"low"为任意的复合业务逻辑名字,这样就可以根据业务的需要灵活地归类的任务了。一般会根据优先级给队列命名(如:high, medium, low).
 如果想要给enqueue传递参数的情况,可以使用enqueue_call方法。在要传递超时参数的情况下:
  1. q = Queue("low", connection = redis_conn)
  2. q.enqueue_call(func=hello, args= ("杨一",),timeout = 30)
3.3  关于worker
   Workers将会从给定的队列中不停的循环读取任务,当所有任务都处理完毕就等待新的work到来。每一个worker在同一时间只处理一个任务。在worker中,是没有并发的。如果你需要并发处理任务,那就需要启动多个worker。
   目前的worker实际上是fork一个子进程来执行具体的任务,也就是说rq不适合windows系统。而且RQ的work是单进程的,如果想要并发执行队列中的任务提高执行效率需要使用threading针对每个任务进行fork线程。
   worker的生命周期有以下几个阶段组成:
   1 启动,载入Python环境
   2 注册,worker注册到系统上,让系统知晓它的存在。
   3 开始监听。从给定的redis队列中取出一个任务。如果所有的队列都是空的且是以突发模式运行的,立即退出。否则,等待新的任务入队。
   4 分配一个子进程。分配的这个子进程在故障安全的上下文中运行实际的任务(调用队列中的任务函数)
   5 处理任务。处理实际的任务。
   6 循环。重复执行步骤3。
四 如何使用
   简单的开发一个deamon 函数,用于后端异步调用,注意任意函数都可以加入队列,必须能够在入队的时候 被程序访问到。
 
  1. #!/usr/bin/env python
  2. #-*- coding:utf-8 -*-
  3. def hello(name):
  4.     print "hello ,%s"%name
  5.     ip='192.168.0.1'
  6.     num=1024
  7.     return name,ip,num
  8. def workat(name):
  9.     print "hello %s ,you r workat youzan.com "%(name)
4.1 构建队列,将任务对象添加到队列里面

  1. >>> from redis import Redis,ConnectionPool
  2. >>> from rq import Queue
  3. >>> pool = ConnectionPool(db=0, host='127.0.0.1', port=6379,
  4. ... password='yangyi')
  5. >>> redis_conn = Redis(connection_pool=pool)
  6. >>> q=Queue(connection=redis_conn)
  7. >>> from rq_test import hello
  8. >>>
  9. >>> result = q.enqueue(hello,'yangyi')
  10. >>> result = q.enqueue(hello,'youzan.com')
先实例化一个Queue类q,然后通过enqueue方法发布任务。第一个参数是执行的函数名,后面是函数执行所需的参数,可以是args也可以是kwargs,案例中是一个字符串。
然后会返回一个Job类的实例,后面会具体介绍Job类的实例具体的api。

4.2启动worker ,从日志上可以看到执行了utils.hello('yangyi') utils.hello('youzan.com') 。当然这个只是简单的调用介绍,生产环境还要写的更加健壮,针对函数执行的结果进行相应的业务逻辑处理。 
  1. root@rac2:~# >python woker.py
  2. 23:44:48 RQ worker u'rq:worker:rac2.3354' started, version 0.6.0
  3. 23:44:48 Cleaning registries for queue: default
  4. 23:44:48
  5. 23:44:48 *** Listening on default...
  6. 23:44:48 default: utils.hello('yangyi') (63879f7c-b453-4405-a262-b9a6b6568b68)
  7. hello ,yangyi
  8. 23:44:48 default: Job OK (63879f7c-b453-4405-a262-b9a6b6568b68)
  9. 23:44:48 Result is kept for 500 seconds
  10. 23:44:48
  11. 23:44:48 *** Listening on default...
  12. 23:45:12 default: utils.hello('youzan.com') (e4e9ed62-c476-45f2-b66a-4b641979e731)
  13. hello ,youzan.com
  14. 23:45:12 default: Job OK (e4e9ed62-c476-45f2-b66a-4b641979e731)
  15. 23:45:12 Result is kept for 500 seconds
需要说明的是其实 worker的启动顺序应该在job放入队列之前,一直监听rq里面是否有具体的任务,当然如果worker晚于job 加入队列启动,job的状态会显示为 queued 状态。
4.3 查看作业执行的情况
当任务加入队列,queue.enqueue()方法返回一个job实例。其定义位于rq.job文件中,可以去查看一下它的API,主要用到的API有:
  1. >>> from rq import job
  2. >>> job = q.enqueue(hello,'youzan.com')
  3. >>> job.get_id() ##获取任务的id ,如果没有指定 ,系统会自动分配一个随机的字符串。
  4. u'17ad0b3a-195e-49d5-8d31-02837ccf5fa6'
  5. >>> job = q.enqueue(hello,'youzan.com')
  6. >>> print job.get_status() ##获取任务的处理状态
  7. finished
  8. >>> step1=q.enqueue(workat,) ##故意不传递参数,让函数执行失败,则获取的状态值是 failed
    >>> print step1.get_status()
    failed
  9. >>> print job.result # 当任务没有执行的时候返回None,否则返回非空值,如果 函数 hello() 有return 的值,会赋值给result
  10. None
  11. 当我们把worker 监听进程停止,然后重新发布任务,查看此时任务的在队列的状态,会显示为 queued
  12. >>> job = q.enqueue(hello,'youzan')
  13. >>> print job.get_status()
  14. queued
  15. >>> print job.to_dict() #把job实例转化成一个字典,我们主要关注状态。
  16. {u'origin': u'default', u'status': u'queued', u'description': u"rq_test.hello('youzan')", u'created_at': '2016-09-06T08:00:40Z', u'enqueued_at': '2016-09-06T08:00:40Z', u'timeout': 180, u'data': '\x80\x02(X\r\x00\x00\x00rq_test.helloq\x01NU\x06youzanq\x02\x85q\x03}q\x04tq\x05.'}
  17. >>> job.cancel() # 取消作业,尽管作业已经被执行,也可以取消
  18. >>> print job.to_dict()
  19. {u'origin': u'default', u'status': u'queued', u'description': u"rq_test.hello('youzan')", u'created_at': '2016-09-06T08:00:40Z', u'enqueued_at': '2016-09-06T08:00:40Z', u'timeout': 180, u'data': '\x80\x02(X\r\x00\x00\x00rq_test.helloq\x01NU\x06youzanq\x02\x85q\x03}q\x04tq\x05.'}
  20. >>> print job.get_status()
  21. queued
  22. >>>
  23. >>> job.delete() # 从redis队列中删除该作业
  24. >>> print job.get_status()
  25. None
  26. >>> print job.to_dict()
  27. {u'origin': u'default', u'description': u"rq_test.hello('youzan')", u'created_at': '2016-09-06T08:00:40Z', u'enqueued_at': '2016-09-06T08:00:40Z', u'timeout': 180, u'data': '\x80\x02(X\r\x00\x00\x00rq_test.helloq\x01NU\x06youzanq\x02\x85q\x03}q\x04tq\x05.'}
五 参考文章
[1] 官方文档  
[2] 翻译 - Python RQ Job 
[3] 翻译 - Python RQ Workers  
[4] 云峰就她了 这位博主写了很多rq相关的实践经验,值得参考。

相关实践学习
基于Redis实现在线游戏积分排行榜
本场景将介绍如何基于Redis数据库实现在线游戏中的游戏玩家积分排行榜功能。
云数据库 Redis 版使用教程
云数据库Redis版是兼容Redis协议标准的、提供持久化的内存数据库服务,基于高可靠双机热备架构及可无缝扩展的集群架构,满足高读写性能场景及容量需弹性变配的业务需求。 产品详情:https://www.aliyun.com/product/kvstore     ------------------------------------------------------------------------- 阿里云数据库体验:数据库上云实战 开发者云会免费提供一台带自建MySQL的源数据库 ECS 实例和一台目标数据库 RDS实例。跟着指引,您可以一步步实现将ECS自建数据库迁移到目标数据库RDS。 点击下方链接,领取免费ECS&RDS资源,30分钟完成数据库上云实战!https://developer.aliyun.com/adc/scenario/51eefbd1894e42f6bb9acacadd3f9121?spm=a2c6h.13788135.J_3257954370.9.4ba85f24utseFl
目录
相关文章
|
8天前
|
机器学习/深度学习 人工智能 算法
基于Python深度学习的眼疾识别系统实现~人工智能+卷积网络算法
眼疾识别系统,本系统使用Python作为主要开发语言,基于TensorFlow搭建卷积神经网络算法,并收集了4种常见的眼疾图像数据集(白内障、糖尿病性视网膜病变、青光眼和正常眼睛) 再使用通过搭建的算法模型对数据集进行训练得到一个识别精度较高的模型,然后保存为为本地h5格式文件。最后使用Django框架搭建了一个Web网页平台可视化操作界面,实现用户上传一张眼疾图片识别其名称。
24 4
基于Python深度学习的眼疾识别系统实现~人工智能+卷积网络算法
|
11天前
|
安全 前端开发 数据库
Python 语言结合 Flask 框架来实现一个基础的代购商品管理、用户下单等功能的简易系统
这是一个使用 Python 和 Flask 框架实现的简易代购系统示例,涵盖商品管理、用户注册登录、订单创建及查看等功能。通过 SQLAlchemy 进行数据库操作,支持添加商品、展示详情、库存管理等。用户可注册登录并下单,系统会检查库存并记录订单。此代码仅为参考,实际应用需进一步完善,如增强安全性、集成支付接口、优化界面等。
|
18天前
|
存储 缓存 监控
局域网屏幕监控系统中的Python数据结构与算法实现
局域网屏幕监控系统用于实时捕获和监控局域网内多台设备的屏幕内容。本文介绍了一种基于Python双端队列(Deque)实现的滑动窗口数据缓存机制,以处理连续的屏幕帧数据流。通过固定长度的窗口,高效增删数据,确保低延迟显示和存储。该算法适用于数据压缩、异常检测等场景,保证系统在高负载下稳定运行。 本文转载自:https://www.vipshare.com
110 66
|
8天前
|
分布式计算 DataWorks 大数据
分布式Python计算服务MaxFrame测评
一文带你入门分布式Python计算服务MaxFrame
55 22
分布式Python计算服务MaxFrame测评
|
3天前
|
分布式计算 DataWorks 数据处理
产品测评 | 上手分布式Python计算服务MaxFrame产品最佳实践
MaxFrame是阿里云自研的分布式计算框架,专为大数据处理设计,提供高效便捷的Python开发体验。其主要功能包括Python编程接口、直接利用MaxCompute资源、与MaxCompute Notebook集成及镜像管理功能。本文基于MaxFrame最佳实践,详细介绍了在DataWorks中使用MaxFrame创建数据源、PyODPS节点和MaxFrame会话的过程,并展示了如何通过MaxFrame实现分布式Pandas处理和大语言模型数据处理。测评反馈指出,虽然MaxFrame具备强大的数据处理能力,但在文档细节和新手友好性方面仍有改进空间。
|
12天前
|
数据采集 人工智能 分布式计算
🚀 MaxFrame 产品深度体验评测:Python 分布式计算的未来
在数据驱动的时代,大数据分析和AI模型训练对数据预处理的效率要求极高。传统的Pandas工具在小数据集下表现出色,但面对大规模数据时力不从心。阿里云推出的Python分布式计算框架MaxFrame,以“Pandas风格”为核心设计理念,旨在降低分布式计算门槛,同时支持超大规模数据处理。MaxFrame不仅保留了Pandas的操作习惯,还通过底层优化实现了高效的分布式调度、内存管理和容错机制,并深度集成阿里云大数据生态。本文将通过实践评测,全面解析MaxFrame的能力与价值,展示其在大数据和AI场景中的卓越表现。
27 4
🚀 MaxFrame 产品深度体验评测:Python 分布式计算的未来
|
6天前
|
SQL 分布式计算 DataWorks
MaxCompute MaxFrame评测 | 分布式Python计算服务MaxFrame(完整操作版)
在当今数字化迅猛发展的时代,数据信息的保存与分析对企业决策至关重要。MaxCompute MaxFrame是阿里云自研的分布式计算框架,支持Python编程接口、兼容Pandas接口并自动进行分布式计算。通过MaxCompute的海量计算资源,企业可以进行大规模数据处理、可视化数据分析及科学计算等任务。本文将详细介绍如何开通MaxCompute和DataWorks服务,并使用MaxFrame进行数据操作。包括创建项目、绑定数据源、编写PyODPS 3节点代码以及执行SQL查询等内容。最后,针对使用过程中遇到的问题提出反馈建议,帮助用户更好地理解和使用MaxFrame。
|
5天前
|
机器学习/深度学习 算法 前端开发
基于Python深度学习果蔬识别系统实现
本项目基于Python和TensorFlow,使用ResNet卷积神经网络模型,对12种常见果蔬(如土豆、苹果等)的图像数据集进行训练,构建了一个高精度的果蔬识别系统。系统通过Django框架搭建Web端可视化界面,用户可上传图片并自动识别果蔬种类。该项目旨在提高农业生产效率,广泛应用于食品安全、智能农业等领域。CNN凭借其强大的特征提取能力,在图像分类任务中表现出色,为实现高效的自动化果蔬识别提供了技术支持。
基于Python深度学习果蔬识别系统实现
|
8天前
|
Python
[oeasy]python057_如何删除print函数_dunder_builtins_系统内建模块
本文介绍了如何删除Python中的`print`函数,并探讨了系统内建模块`__builtins__`的作用。主要内容包括: 1. **回忆上次内容**:上次提到使用下划线避免命名冲突。 2. **双下划线变量**:解释了双下划线(如`__name__`、`__doc__`、`__builtins__`)是系统定义的标识符,具有特殊含义。
20 3
|
8天前
|
分布式计算 数据处理 MaxCompute
云产品评测|分布式Python计算服务MaxFrame
云产品评测|分布式Python计算服务MaxFrame
35 2