/1 前言/
大家好，我是吴老板。用Celery 官方的话来说，Celery 是一个非常优秀的分布式队列，可应用于分布式共享中间队列和定时任务等等。
/2 版本的差异/
Celery 有很多个版本，各版本之间的差异可谓不小，比如最新的 Celery6.0 版本在稳定性远不如 Celery4.0，所以在使用不同版本的时候，系统给到我们的反馈可能并不能如我们所愿。
/3 服务/
在 windows 下挂在 Celery 服务有时候会出现不稳定的情况(unix中暂时未发现这种情况)，比如在执行定时任务的时候，过了一段时间之后，Celery 出现了假死状态，以至于不能按照我们指定的时间点去执行任务。
这些任务只是加入到待运行队列中(堆积在 Redis 中)，只能人为重启 Celery 服务之后才能将堆积的任务释放出来运行。
这样一来，第一是定时任务在指定时间点没有正常运行，其二是在其他时间运行了这些任务，很可能会产生更新数据不及时，时间节点混乱的问题，不仅达不到业务需求，还会反受其害。
/4 设置心跳/
为了解决 Celery 在 windows 中的这种弊端，可以为 Celery 任务队列设置一个心跳时间，比如每一分钟或者每五分钟向 Redis 数据库发送一次数据以保证队列始终是活跃的状态，这样只要你的电脑不关机并保持网络畅通(如果是远程 Redis)，Celery 任务队列服务就不会出现假死状态。
/5 举个栗子/
我总是很喜欢用示例来说话，前些时间在对某平台的商家后台进行数据采集的时候，为了使用时能自动获取该网站的 cookie ，
用Pyppeteer 写了一个自动化登陆的脚本，和往常一样仍在 Celery 队列中并迅速的启动服务。
脚本是这样的(非常接近实际的伪代码，没办法，保命要紧)
1# -*- coding: utf-8 -*- 2from db.redisCurd import RedisQueue 3import asyncio 4import random 5import tkinter 6from pyppeteer.launcher import launch 7from platLogin.config import USERNAME, PASSWORD, LOGIN_URL 8 9class Login(): 10 def __init__(self, shopId): 11 self.shopId = shopId 12 self.RedisQueue = RedisQueue("cookie") 13 14 def screen_size(self): 15 tk = tkinter.Tk() 16 width = tk.winfo_screenwidth() 17 height = tk.winfo_screenheight() 18 tk.quit() 19 return {'width': width, 'height': height} 20 21 async def login(self, username, password, url): 22 browser = await launch( 23 { 24 'headless': False, 25 'dumpio': True 26 }, 27 args=['--no-sandbox', '--disable-infobars', '--user-data-dir=./userData'], 28 ) 29 page = await browser.newPage() # 启动新的浏览器页面 30 31 try: 32 await page.setViewport(viewport=self.screen_size()) 33 await page.setJavaScriptEnabled(enabled=True) # 启用js 34 await page.setUserAgent( 35 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.36 Edge/16.16299' 36 ) 37 await self.page_evaluate(page) 38 await page.goto(url) 39 await asyncio.sleep(2) 40 # 输入用户名，密码 41 await page.evaluate(f'document.querySelector("#userName").value=""') 42 await page.type('#userName', username, {'delay': self.input_time_random() - 50}) # delay是限制输入的时间 43 await page.evaluate('document.querySelector("#passWord").value=""') 44 await page.type('#passWord', password, {'delay': self.input_time_random()}) 45 await page.waitFor(6000) 46 47 loginImgVcode = await page.waitForSelector('#checkCode') 48 await loginImgVcode.screenshot({'path': './loginImg.png'}) 49 await page.waitFor(6000) 50 51 res = use_cjy("./loginImg.png") 52 pic_str = res.get("pic_str") if res.get("err_str") == "OK" else "1234" 53 54 await page.waitFor(6000) 55 await page.type('#checkWord', pic_str, {'delay': self.input_time_random() - 50}) 56 await page.waitFor(6000) 57 58 await page.click('#subMit') 59 await page.waitFor(6000) 60 await asyncio.sleep(2) 61 await self.get_cookie(page) 62 await page.waitFor(3000) 63 await self.page_close(browser) 64 return {'code': 200, 'msg': '登陆成功'} 65 except: 66 return {'code': -1, 'msg': '出错'} 67 68 finally: 69 await page.waitFor(3000) 70 await self.page_close(browser) 71 72 # 获取登录后cookie 73 async def get_cookie(self, page): 74 cookies_list = await page.cookies() 75 cookies = '' 76 for cookie in cookies_list: 77 str_cookie = '{0}={1}; ' 78 str_cookie = str_cookie.format(cookie.get('name'), cookie.get('value')) 79 cookies += str_cookie 80 # 将cookie 放入 cookie 池 81 self.RedisQueue.put_hash(self.shopId, cookies) 82 return cookies 83 84 async def page_evaluate(self, page): 85 await page.evaluate('''() =>{ Object.defineProperties(navigator,{ webdriver:{ get: () => undefined } }) }''') 86 await page.evaluate('''() =>{ window.navigator.chrome = { runtime: {}, }; }''') 87 await page.evaluate( 88 '''() =>{ Object.defineProperty(navigator, 'languages', { get: () => ['en-US', 'en'] }); }''') 89 await page.evaluate( 90 '''() =>{ Object.defineProperty(navigator, 'plugins', { get: () => [1, 2, 3, 4, 5,6], }); }''') 91 await page.waitFor(3000) 92 93 async def page_close(self, browser): 94 for _page in await browser.pages(): 95 await _page.close() 96 await browser.close() 97 98 def input_time_random(self): 99 return random.randint(100, 151) 100 101 def run(self, username=USERNAME, password=PASSWORD, url=LOGIN_URL): 102 loop = asyncio.get_event_loop() 103 i_future = asyncio.ensure_future(self.login(username, password, url)) 104 loop.run_until_complete(i_future) 105 return i_future.result() 106 107 108if __name__ == '__main__': 109 Z = Login(shopId="001") 110 Z.run()
Celery 任务文件是这样的
1 2# -*- coding: utf-8 -*- 3from __future__ import absolute_import 4import os 5import sys 6import time 7from db.redisCurd import RedisQueue 8from send_msg.weinxin import Send_msg 9base_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) 10sys.path.append(base_dir) 11from logger.logger import log_v 12from celery import Task 13from platLogin.login import Login # 登陆类 14from celery import Celery 15 16randomQueue = RedisQueue("cookie") 17 18celery_app = Celery('task') 19celery_app.config_from_object('celeryConfig') 20 21S = Send_msg() 22 23dl_dict = { 24 'demo': { 25 'cookie': '', 26 'loginClass': 'Login', 27 } 28} 29 30# todo 这是三种运行的状态 31class task_status(Task): 32 def on_success(self, retval, task_id, args, kwargs): 33 log_v.info('任务信息 -> id:{} , arg:{} , successful ..... Done'.format(task_id, args)) 34 35 def on_failure(self, exc, task_id, args, kwargs, einfo): 36 log_v.error('task id:{} , arg:{} , failed ! error : {}'.format(task_id, args, exc)) 37 38 def on_retry(self, exc, task_id, args, kwargs, einfo): 39 log_v.warning('task id:{} , arg:{} , retry ! info: {}'.format(task_id, args, exc)) 40 41 42# todo 随便找个hash key作为轮询对象, celery在win10系统可能不太稳定,有时候会有连接断开的情况 43@celery_app.task(base=task_status) 44def get_cookie_status(platName="demo"): 45 try: 46 # log_v.debug(f'[+] 轮询 {platName} 定时器启动 ..... Done') 47 randomQueue.get_hash(platName).decode() 48 log_v.debug(f'[+] 轮询 {platName} 成功 ..... Done') 49 return "Erp 轮询成功" 50 except: 51 return "Erp 轮询失败" 52 53 54@celery_app.task(base=task_status) 55def set_plat_cookie(platName="demo", shopId=None): 56 log_v.debug(f"[+] {platName} 正在登陆") 57 core = eval(dl_dict[platName]['loginClass'])(shopId=shopId) 58 result = core.run() 59 return result
Celery 配置文件是这样的
1from __future__ import absolute_import 2import datetime 3from kombu import Exchange, Queue 4from celery.schedules import crontab 5from urllib import parse 6 7BROKER_URL = f'redis://root:{parse.quote("你的不规则密码")}@主机:6379/15' 8 9# 导入任务，如tasks.py 10CELERY_IMPORTS = ('monitor.tasks',) 11 12# 列化任务载荷的默认的序列化方式 13CELERY_TASK_SERIALIZER = 'json' 14 15# 结果序列化方式 16CELERY_RESULT_SERIALIZER = 'json' 17CELERY_ACCEPT_CONTENT = ['json'] 18 19CELERY_TIMEZONE = 'Asia/Shanghai' # 指定时区，不指定默认为 'UTC' 20# CELERY_TIMEZONE='UTC' 21 22CELERYBEAT_SCHEDULE = { 23 'add-every-60-seconds': { 24 'task': 'tasks.get_cookie_status', 25 'schedule': datetime.timedelta(minutes=1), # 每 1 分钟执行一次 26 'args': () # 任务函数参数 27 }, 28}
启动服务
1celery -A tasks beat -l INFO 2celery -A tasks worker -l INFO -c 2
以 2 个线程启动消费者队列服务并启用定时任务，当发现当前平台的 cookie 不可用时，我会向 Celery 发送一个信号(就是调用了前面的set_plat_cookie 这个方法)，消费者得到这个任务这个就会执行自动化脚本以获取 cookie 并储存在 Redis 中，使用时在从 Redis 中获取就能正常请求到该平台的数据。
在空闲时间，Celery中的 get_cookie_status 方法会每隔一分钟向 Redis 请求数据，这就是我们设置的 1分钟心跳。
这样不管我们的 Celery 是否是后台启动，都不会出现假死、卡死的状态，则万事大吉矣！！
/6 总结/
本文为了解决 Celery 在 windows 中的这种弊端，为 Celery 任务队列设置一个心跳时间，比如每一分钟或者每五分钟向 Redis 数据库发送一次数据以保证队列始终是活跃的状态，这样只要你的电脑不关机并保持网络畅通(如果是远程 Redis)，Celery 任务队列服务都不会出现假死、卡死的状态。
-------------------********************************** End **********-------------**-----********-**********************************
往期精彩文章推荐：
欢迎各位大佬点击链接加入群聊【helloworld开发者社区】：https://jq.qq.com/?_wv=1027&k=mBlk6nzX进群交流IT技术热点。
本文转自 https://mp.weixin.qq.com/s/qi9q8qwcySEQbb51aEPXTQ，如有侵权，请联系删除。