手把手教你在Windows下设置分布式队列Celery的心跳轮询

手把手教你在Windows下设置分布式队列Celery的心跳轮询

作者： 吴老板 2021-03-05 08:52:00

系统

Windows

分布式 大家好，我是吴老板。用Celery 官方的话来说，Celery 是一个非常优秀的分布式队列，可应用于分布式共享中间队列和定时任务等等。

10多年的章贡网站建设经验，针对设计、前端、开发、售后、文案、推广等六对一服务，响应快，48小时及时工作处理。成都营销网站建设的优势是能够根据用户设备显示端的尺寸不同，自动调整章贡建站的显示方式，使网站能够适用不同显示终端，在浏览器中调整网站的宽度，无论在任何一种浏览器上浏览网站，都能展现优雅布局与设计，从而大程度地提升浏览体验。成都创新互联从事“章贡网站设计”,“章贡网站推广”以来，每个客户项目都认真落实执行。

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 -*- 
    2. 
   
  
  
  2. from db.redisCurd import RedisQueue 
    3. 
   
  
  
  3. import asyncio 
    4. 
   
  
  
  4. import random 
    5. 
   
  
  
  5. import tkinter 
    6. 
   
  
  
  6. from pyppeteer.launcher import launch 
    7. 
   
  
  
  7. from platLogin.config import USERNAME, PASSWORD, LOGIN_URL 
    8. 
   
  
  
  8.  
    9. 
   
  
  
  9. class Login(): 
    10. 
   
  
  
  10.     def __init__(self, shopId): 
    11. 
   
  
  
  11.         self.shopId = shopId 
    12. 
   
  
  
  12.         self.RedisQueue = RedisQueue("cookie") 
    13. 
   
  
  
  13.  
    14. 
   
  
  
  14.     def screen_size(self): 
    15. 
   
  
  
  15.         tk = tkinter.Tk() 
    16. 
   
  
  
  16.         width = tk.winfo_screenwidth() 
    17. 
   
  
  
  17.         height = tk.winfo_screenheight() 
    18. 
   
  
  
  18.         tk.quit() 
    19. 
   
  
  
  19.         return {'width': width, 'height': height} 
    20. 
   
  
  
  20.  
    21. 
   
  
  
  21.     async def login(self, username, password, url): 
    22. 
   
  
  
  22.         browser = await launch( 
    23. 
   
  
  
  23.             { 
    24. 
   
  
  
  24.                 'headless': False, 
    25. 
   
  
  
  25.                 'dumpio': True 
    26. 
   
  
  
  26.             }, 
    27. 
   
  
  
  27.             args=['--no-sandbox', '--disable-infobars', '--user-data-dir=./userData'], 
    28. 
   
  
  
  28.         ) 
    29. 
   
  
  
  29.         page = await browser.newPage()  # 启动新的浏览器页面 
    30. 
   
  
  
  30.  
    31. 
   
  
  
  31.         try: 
    32. 
   
  
  
  32.             await page.setViewport(viewport=self.screen_size()) 
    33. 
   
  
  
  33.             await page.setJavaScriptEnabled(enabled=True)  # 启用js 
    34. 
   
  
  
  34.             await page.setUserAgent( 
    35. 
   
  
  
  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. 
   
  
  
  36.             ) 
    37. 
   
  
  
  37.             await self.page_evaluate(page) 
    38. 
   
  
  
  38.             await page.goto(url) 
    39. 
   
  
  
  39.             await asyncio.sleep(2) 
    40. 
   
  
  
  40.             # 输入用户名，密码 
    41. 
   
  
  
  41.             await page.evaluate(f'document.querySelector("#userName").value=""') 
    42. 
   
  
  
  42.             await page.type('#userName', username, {'delay': self.input_time_random() - 50})  # delay是限制输入的时间 
    43. 
   
  
  
  43.             await page.evaluate('document.querySelector("#passWord").value=""') 
    44. 
   
  
  
  44.             await page.type('#passWord', password, {'delay': self.input_time_random()}) 
    45. 
   
  
  
  45.             await page.waitFor(6000) 
    46. 
   
  
  
  46.  
    47. 
   
  
  
  47.             loginImgVcode = await page.waitForSelector('#checkCode')   
    48. 
   
  
  
  48.             await loginImgVcode.screenshot({'path': './loginImg.png'}) 
    49. 
   
  
  
  49.             await page.waitFor(6000) 
    50. 
   
  
  
  50.  
    51. 
   
  
  
  51.             res = use_cjy("./loginImg.png") 
    52. 
   
  
  
  52.             pic_str = res.get("pic_str") if res.get("err_str") == "OK" else "1234" 
    53. 
   
  
  
  53.  
    54. 
   
  
  
  54.             await page.waitFor(6000) 
    55. 
   
  
  
  55.             await page.type('#checkWord', pic_str, {'delay': self.input_time_random() - 50}) 
    56. 
   
  
  
  56.             await page.waitFor(6000) 
    57. 
   
  
  
  57.  
    58. 
   
  
  
  58.             await page.click('#subMit') 
    59. 
   
  
  
  59.             await page.waitFor(6000) 
    60. 
   
  
  
  60.             await asyncio.sleep(2) 
    61. 
   
  
  
  61.             await self.get_cookie(page) 
    62. 
   
  
  
  62.             await page.waitFor(3000) 
    63. 
   
  
  
  63.             await self.page_close(browser) 
    64. 
   
  
  
  64.             return {'code': 200, 'msg': '登陆成功'} 
    65. 
   
  
  
  65.         except: 
    66. 
   
  
  
  66.             return {'code': -1, 'msg': '出错'} 
    67. 
   
  
  
  67.  
    68. 
   
  
  
  68.         finally: 
    69. 
   
  
  
  69.             await page.waitFor(3000) 
    70. 
   
  
  
  70.             await self.page_close(browser) 
    71. 
   
  
  
  71.  
    72. 
   
  
  
  72.     # 获取登录后cookie 
    73. 
   
  
  
  73.     async def get_cookie(self, page): 
    74. 
   
  
  
  74.         cookies_list = await page.cookies() 
    75. 
   
  
  
  75.         cookies = '' 
    76. 
   
  
  
  76.         for cookie in cookies_list: 
    77. 
   
  
  
  77.             str_cookie = '{0}={1}; ' 
    78. 
   
  
  
  78.             str_cookie = str_cookie.format(cookie.get('name'), cookie.get('value')) 
    79. 
   
  
  
  79.             cookies += str_cookie 
    80. 
   
  
  
  80.         # 将cookie 放入 cookie 池 
    81. 
   
  
  
  81.         self.RedisQueue.put_hash(self.shopId, cookies) 
    82. 
   
  
  
  82.         return cookies 
    83. 
   
  
  
  83.  
    84. 
   
  
  
  84.     async def page_evaluate(self, page): 
    85. 
   
  
  
  85.         await page.evaluate('''() =>{ Object.defineProperties(navigator,{ webdriver:{ get: () => undefined } }) }''') 
    86. 
   
  
  
  86.         await page.evaluate('''() =>{ window.navigator.chrome = { runtime: {},  }; }''') 
    87. 
   
  
  
  87.         await page.evaluate( 
    88. 
   
  
  
  88.             '''() =>{ Object.defineProperty(navigator, 'languages', { get: () => ['en-US', 'en'] }); }''') 
    89. 
   
  
  
  89.         await page.evaluate( 
    90. 
   
  
  
  90.             '''() =>{ Object.defineProperty(navigator, 'plugins', { get: () => [1, 2, 3, 4, 5,6], }); }''') 
    91. 
   
  
  
  91.         await page.waitFor(3000) 
    92. 
   
  
  
  92.  
    93. 
   
  
  
  93.     async def page_close(self, browser): 
    94. 
   
  
  
  94.         for _page in await browser.pages(): 
    95. 
   
  
  
  95.             await _page.close() 
    96. 
   
  
  
  96.         await browser.close() 
    97. 
   
  
  
  97.  
    98. 
   
  
  
  98.     def input_time_random(self): 
    99. 
   
  
  
  99.         return random.randint(100, 151) 
    100. 
   
  
  
  100.  
    101. 
   
  
  
  101.     def run(self, username=USERNAME, password=PASSWORD, url=LOGIN_URL): 
    102. 
   
  
  
  102.         loop = asyncio.get_event_loop() 
    103. 
   
  
  
  103.         i_future = asyncio.ensure_future(self.login(username, password, url)) 
    104. 
   
  
  
  104.         loop.run_until_complete(i_future) 
    105. 
   
  
  
  105.         return i_future.result() 
    106. 
   
  
  
  106.  
    107. 
   
  
  
  107.  
    108. 
   
  
  
  108. if __name__ == '__main__': 
    109. 
   
  
  
  109.     Z = Login(shopId="001") 
    110. 
   
  
  
  110.     Z.run() 
    111.

Celery 任务文件是这样的

  
 
 
    
   
  
  
  1. # -*- coding: utf-8 -*- 
    2. 
   
  
  
  2. from __future__ import absolute_import 
    3. 
   
  
  
  3. import os 
    4. 
   
  
  
  4. import sys 
    5. 
   
  
  
  5. import time 
    6. 
   
  
  
  6. from db.redisCurd import RedisQueue 
    7. 
   
  
  
  7. from send_msg.weinxin import Send_msg 
    8. 
   
  
  
  8. base_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) 
    9. 
   
  
  
  9. sys.path.append(base_dir) 
    10. 
   
  
  
  10. from logger.logger import log_v 
    11. 
   
  
  
  11. from celery import Task 
    12. 
   
  
  
  12. from platLogin.login import Login  # 登陆类 
    13. 
   
  
  
  13. from celery import Celery 
    14. 
   
  
  
  14.  
    15. 
   
  
  
  15. randomQueue = RedisQueue("cookie") 
    16. 
   
  
  
  16.  
    17. 
   
  
  
  17. celery_app = Celery('task') 
    18. 
   
  
  
  18. celery_app.config_from_object('celeryConfig') 
    19. 
   
  
  
  19.  
    20. 
   
  
  
  20. S = Send_msg() 
    21. 
   
  
  
  21.  
    22. 
   
  
  
  22. dl_dict = { 
    23. 
   
  
  
  23.     'demo': { 
    24. 
   
  
  
  24.         'cookie': '', 
    25. 
   
  
  
  25.         'loginClass': 'Login', 
    26. 
   
  
  
  26.     } 
    27. 
   
  
  
  27. } 
    28. 
   
  
  
  28.  
    29. 
   
  
  
  29. # todo 这是三种运行的状态 
    30. 
   
  
  
  30. class task_status(Task): 
    31. 
   
  
  
  31.     def on_success(self, retval, task_id, args, kwargs):  
    32. 
   
  
  
  32.         log_v.info('任务信息 -> id:{} , arg:{} , successful ..... Done'.format(task_id, args)) 
    33. 
   
  
  
  33.  
    34. 
   
  
  
  34.     def on_failure(self, exc, task_id, args, kwargs, einfo):   
    35. 
   
  
  
  35.         log_v.error('task id:{} , arg:{} , failed ! error : {}'.format(task_id, args, exc)) 
    36. 
   
  
  
  36.  
    37. 
   
  
  
  37.     def on_retry(self, exc, task_id, args, kwargs, einfo):  
    38. 
   
  
  
  38.         log_v.warning('task id:{} , arg:{} , retry !  info: {}'.format(task_id, args, exc)) 
    39. 
   
  
  
  39.  
    40. 
   
  
  
  40.  
    41. 
   
  
  
  41. # todo 随便找个hash key作为轮询对象, celery在win10系统可能不太稳定,有时候会有连接断开的情况 
    42. 
   
  
  
  42. @celery_app.task(base=task_status) 
    43. 
   
  
  
  43. def get_cookie_status(platName="demo"): 
    44. 
   
  
  
  44.     try: 
    45. 
   
  
  
  45.         # log_v.debug(f'[+] 轮询 {platName} 定时器启动 ..... Done') 
    46. 
   
  
  
  46.         randomQueue.get_hash(platName).decode() 
    47. 
   
  
  
  47.         log_v.debug(f'[+] 轮询 {platName} 成功 ..... Done') 
    48. 
   
  
  
  48.         return "Erp 轮询成功" 
    49. 
   
  
  
  49.     except: 
    50. 
   
  
  
  50.         return "Erp 轮询失败" 
    51. 
   
  
  
  51.  
    52. 
   
  
  
  52.  
    53. 
   
  
  
  53. @celery_app.task(base=task_status) 
    54. 
   
  
  
  54. def set_plat_cookie(platName="demo", shopId=None): 
    55. 
   
  
  
  55.     log_v.debug(f"[+] {platName} 正在登陆") 
    56. 
   
  
  
  56.     core = eval(dl_dict[platName]['loginClass'])(shopId=shopId) 
    57. 
   
  
  
  57.     result = core.run() 
    58. 
   
  
  
  58.     return result 
    59.

Celery 配置文件是这样的

  
 
 
    
   
  
  
  1. from __future__ import absolute_import 
    2. 
   
  
  
  2. import datetime 
    3. 
   
  
  
  3. from kombu import Exchange, Queue 
    4. 
   
  
  
  4. from celery.schedules import crontab 
    5. 
   
  
  
  5. from urllib import parse 
    6. 
   
  
  
  6.  
    7. 
   
  
  
  7. BROKER_URL = f'redis://root:{parse.quote("你的不规则密码")}@主机:6379/15' 
    8. 
   
  
  
  8.  
    9. 
   
  
  
  9. # 导入任务，如tasks.py 
    10. 
   
  
  
  10. CELERY_IMPORTS = ('monitor.tasks',) 
    11. 
   
  
  
  11.  
    12. 
   
  
  
  12. # 列化任务载荷的默认的序列化方式 
    13. 
   
  
  
  13. CELERY_TASK_SERIALIZER = 'json' 
    14. 
   
  
  
  14.  
    15. 
   
  
  
  15. # 结果序列化方式 
    16. 
   
  
  
  16. CELERY_RESULT_SERIALIZER = 'json' 
    17. 
   
  
  
  17. CELERY_ACCEPT_CONTENT = ['json'] 
    18. 
   
  
  
  18.  
    19. 
   
  
  
  19. CELERY_TIMEZONE = 'Asia/Shanghai'  # 指定时区，不指定默认为 'UTC' 
    20. 
   
  
  
  20. # CELERY_TIMEZONE='UTC' 
    21. 
   
  
  
  21.  
    22. 
   
  
  
  22. CELERYBEAT_SCHEDULE = { 
    23. 
   
  
  
  23.     'add-every-60-seconds': { 
    24. 
   
  
  
  24.         'task': 'tasks.get_cookie_status', 
    25. 
   
  
  
  25.         'schedule': datetime.timedelta(minutes=1),  # 每 1 分钟执行一次 
    26. 
   
  
  
  26.         'args': ()  # 任务函数参数 
    27. 
   
  
  
  27.     }, 
    28. 
   
  
  
  28. } 
    29.

启动服务

  
 
 
    
   
  
  
  1. celery -A tasks beat -l INFO 
    2. 
   
  
  
  2. celery -A tasks worker -l INFO -c 2 
    3.

以 2 个线程启动消费者队列服务并启用定时任务，当发现当前平台的 cookie 不可用时，我会向 Celery 发送一个信号(就是调用了前面的set_plat_cookie 这个方法)，消费者得到这个任务这个就会执行自动化脚本以获取 cookie 并储存在 Redis 中，使用时在从 Redis 中获取就能正常请求到该平台的数据。

在空闲时间，Celery中的 get_cookie_status 方法会每隔一分钟向 Redis 请求数据，这就是我们设置的 1分钟心跳。

这样不管我们的 Celery 是否是后台启动，都不会出现假死、卡死的状态，则万事大吉矣!!

6 总结

本文为了解决 Celery 在 windows 中的这种弊端，为 Celery 任务队列设置一个心跳时间，比如每一分钟或者每五分钟向 Redis 数据库发送一次数据以保证队列始终是活跃的状态，这样只要你的电脑不关机并保持网络畅通(如果是远程 Redis)，Celery 任务队列服务都不会出现假死、卡死的状态。

文章名称：手把手教你在Windows下设置分布式队列Celery的心跳轮询
浏览路径：http://www.shufengxianlan.com/qtweb/news10/199860.html

网站建设、网络推广公司-创新互联，是专注品牌与效果的网站制作，网络营销seo公司；服务项目有等

广告

声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源： 创新互联

猜你还喜欢下面的内容

品牌网站制作知识

分类信息网

成都恒翔机电维修　　 　成都企业网站维护　　 　php空间　　 　新津网站运维　　 　橡塑保温管　　 　南宁旺客科技　　 　成都光华机房　　 　手机网站建设　　 　网站营销推广　　 　普特斯腐植酸　　 　成都小程序商城　　 　和正源建设　　 　VI设计　　 　网站建设公司　　 　混凝土搅拌罐车　　 　app软件开发　　 　江油网站建设　　 　达州博佩科技　　 　新华建筑装饰　　 　证企宝企业