芹菜打败SpawnPoolWorker

分类：编程问答

如何解决芹菜打败SpawnPoolWorker

我有一个函数，Celery Beat每5秒调用一次，并将其附加到全局变量中。 我希望我的函数每5秒钟将元素添加到全局变量，但每20秒钟就会添加一次。

这是task.py：

# tasks.py
from celery import shared_task
from .celeryapp import app
from . import cfg

@shared_task
def update_a_global_list():
    try:
        if cfg.flag:
            cfg.init()
        l = ['first']
        cfg.my_global_var.append(l)
        print("my_global_var: " + str(cfg.my_global_var))
    except Exception as e:
        print(e)

全局变量位于一个名为cfg.py的文件中：

# cfg.py
global my_global_var
global flag
flag = True


def init():
    global flag
    flag = False
    global my_global_var
    my_global_var = []
    print('Initialize Step')

该项目的celery配置位于celeryapp.py中：

# celeryapp.py
from __future__ import absolute_import
from celery import Celery
from celery.schedules import crontab

import os

os.environ.setdefault('FORKED_BY_MULTIPROCESSING','1')

app = Celery('tasks',broker='amqp://shahab_user:pass1234@localhost:5672/shahab_vhost',backend='rpc://')

app.conf.beat_schedule = {
    'every-5-seconds': {
        'task': 'send_requests.tasks.update_a_global_list','schedule': 5,},}

当我运行命令时：

芹菜-任务工作者-l信息

在一个终端中运行命令：

芹菜-send_requests.celeryapp节拍-l信息

在另一个终端中，我在工作程序终端中看到以下日志：

[2020-10-07 16:39:48,545: INFO/MainProcess] Received task: 
send_requests.tasks.update_a_global_list[c24168e6-49df-4188-be2e-8ea05a563f2a]
[2020-10-07 16:39:48,545: WARNING/SpawnPoolWorker-1] Initialize Step                    <===
[2020-10-07 16:39:48,561: WARNING/SpawnPoolWorker-1] my_global_var: [['first']]         <===
[2020-10-07 16:39:48,670: INFO/SpawnPoolWorker-1] Task 
send_requests.tasks.update_a_global_list[c24168e6-49df-4188-be2e-8ea05a563f2a] succeeded in 0.125s :None
[2020-10-07 16:39:53,440: INFO/MainProcess] Received task: 
send_requests.tasks.update_a_global_list[ce0ec733-be6d-4640-950f-a2f47ecf1693]
[2020-10-07 16:39:53,440: WARNING/SpawnPoolWorker-2] Initialize Step                    <===
[2020-10-07 16:39:53,440: WARNING/SpawnPoolWorker-2] my_global_var: [['first']]         <===
[2020-10-07 16:39:53,547: INFO/SpawnPoolWorker-2] Task 
send_requests.tasks.update_a_global_list[ce0ec733-be6d-4640-950f-a2f47ecf1693] succeeded in 0.0940000000409782s: None
[2020-10-07 16:39:58,450: INFO/MainProcess] Received task: 
send_requests.tasks.update_a_global_list[fae64a0a-8132-4d88-bcd1-cc59d7e73794]
[2020-10-07 16:39:58,453: WARNING/SpawnPoolWorker-3] Initialize Step                    <===
[2020-10-07 16:39:58,454: WARNING/SpawnPoolWorker-3] my_global_var: [['first']]         <===
[2020-10-07 16:39:58,532: INFO/SpawnPoolWorker-3] Task 
send_requests.tasks.update_a_global_list[fae64a0a-8132-4d88-bcd1-cc59d7e73794] succeeded in 0.0779999999795109s: None
[2020-10-07 16:40:03,453: INFO/MainProcess] Received task: 
send_requests.tasks.update_a_global_list[86d92236-aaf4-4c62-81d6-2679eed287b2]
[2020-10-07 16:40:03,453: WARNING/SpawnPoolWorker-4] Initialize Step                    <===
[2020-10-07 16:40:03,453: WARNING/SpawnPoolWorker-4] my_global_var: [['first']]         <===
[2020-10-07 16:40:03,533: INFO/SpawnPoolWorker-4] Task 
send_requests.tasks.update_a_global_list[86d92236-aaf4-4c62-81d6-2679eed287b2] succeeded in 0.0779999999795109s: None
[2020-10-07 16:40:08,467: INFO/MainProcess] Received task: 
send_requests.tasks.update_a_global_list[1c9fb4e4-a364-4079-98fa-f6deeb3ad638]
[2020-10-07 16:40:08,469: WARNING/SpawnPoolWorker-1] my_global_var: [['first'],['first']]         <===
[2020-10-07 16:40:08,472: INFO/SpawnPoolWorker-1] Task 
send_requests.tasks.update_a_global_list[1c9fb4e4-a364-4079-98fa-f6deeb3ad638] succeeded in 0.015999999945051968s: None
[2020-10-07 16:40:13,463: INFO/MainProcess] Received task: 
send_requests.tasks.update_a_global_list[654d9da4-2c13-46c3-88fe-7b4e73edc8d0]
[2020-10-07 16:40:13,465: WARNING/SpawnPoolWorker-2] my_global_var: [['first'],['first']]         <===
[2020-10-07 16:40:13,468: INFO/SpawnPoolWorker-2] Task 
send_requests.tasks.update_a_global_list[654d9da4-2c13-46c3-88fe-7b4e73edc8d0] succeeded in 0.0s:None
[2020-10-07 16:40:18,465: INFO/MainProcess] Received task: 
send_requests.tasks.update_a_global_list[8d777ade-870b-4133-9f26-8467e4ca4ba7]
[2020-10-07 16:40:18,467: WARNING/SpawnPoolWorker-3] my_global_var: [['first'],['first']]         <===
[2020-10-07 16:40:18,470: INFO/SpawnPoolWorker-3] Task 
send_requests.tasks.update_a_global_list[8d777ade-870b-4133-9f26-8467e4ca4ba7] succeeded in 0.0s:None
[2020-10-07 16:40:23,470: INFO/MainProcess] Received task: 
send_requests.tasks.update_a_global_list[dc8bfa49-0f3e-48bc-9c4d-33705048ffce]
[2020-10-07 16:40:23,473: WARNING/SpawnPoolWorker-4] my_global_var: [['first'],['first']]         <===
[2020-10-07 16:40:23,476: INFO/SpawnPoolWorker-4] Task 
send_requests.tasks.update_a_global_list[dc8bfa49-0f3e-48bc-9c4d-33705048ffce] succeeded in 0.0s:None
[2020-10-07 16:40:28,465: INFO/MainProcess] Received task: 
send_requests.tasks.update_a_global_list[bad53432-03cb-4843-b71b-105c8d83c971]
[2020-10-07 16:40:28,467: WARNING/SpawnPoolWorker-1] my_global_var: [['first'],['first'],['first']]     <===
[2020-10-07 16:40:28,470: INFO/SpawnPoolWorker-1] Task 
send_requests.tasks.update_a_global_list[bad53432-03cb-4843-b71b-105c8d83c971] succeeded in 0.0s:None

为什么“初始化步骤”运行超过一次？

为什么我有不同的SpawnPoolWorker，却没有告诉我我的期望？

感谢您的帮助。

[编辑]： 根据@DejanLekic的答案，我还使用了Django缓存。但是我得到了相同的结果。

这次，我以这种方式编写程序：

@shared_task
def update_global_list():
    try:
        test = []
        l = ['first']
        cached_object = cache.get('my_global_var')
        if cached_object is None:
            cache.set('my_global_var',test)
            cached_object = cache.get('my_global_var')
        cached_object.append(l)
        cache.set('my_global_var',cached_object)
        print("my_global_var : " + str(cache.get('my_global_var')))
    except Exception as e:
        print(e)

我得到了这些结果：

[2020-10-09 14:11:45,903: INFO/MainProcess] Received task: 
send_requests.tasks.update_global_list[93264e6e-3cd0-4068-ac9e-98f366cdfd51]
[2020-10-09 14:11:45,907: WARNING/SpawnPoolWorker-1] my_global_var : 
[['first']]
[2020-10-09 14:11:45,963: INFO/SpawnPoolWorker-1] Task 
send_requests.tasks.update_global_list[93264e6e-3cd0-4068-ac9e-98f366cdfd51] 
succeeded in 0.0470000000204891s: None
[2020-10-09 14:11:50,835: INFO/MainProcess] Received task: 
send_requests.tasks.update_global_list[0673a0a8-8b79-4325-a9aa-015061f76166]
[2020-10-09 14:11:50,838: WARNING/SpawnPoolWorker-2] my_global_var : 
[['first']]
[2020-10-09 14:11:50,887: INFO/SpawnPoolWorker-2] Task 
send_requests.tasks.update_global_list[0673a0a8-8b79-4325-a9aa-015061f76166] 
succeeded in 0.0470000000204891s: None
[2020-10-09 14:11:55,830: INFO/MainProcess] Received task: 
send_requests.tasks.update_global_list[2c58b9b3-ae37-4b8a-b28f-8e4dce2952aa]
[2020-10-09 14:11:55,834: WARNING/SpawnPoolWorker-3] my_global_var : 
[['first']]
[2020-10-09 14:11:55,884: INFO/SpawnPoolWorker-3] Task 
send_requests.tasks.update_global_list[2c58b9b3-ae37-4b8a-b28f-8e4dce2952aa] 
succeeded in 0.0470000000204891s: None
[2020-10-09 14:12:00,829: INFO/MainProcess] Received task: 
send_requests.tasks.update_global_list[2fc7a1a9-27e7-41dc-936d-a017d2a283bc]
[2020-10-09 14:12:00,833: WARNING/SpawnPoolWorker-4] my_global_var : 
[['first']]
[2020-10-09 14:12:00,951: INFO/SpawnPoolWorker-4] Task 
send_requests.tasks.update_global_list[2fc7a1a9-27e7-41dc-936d-a017d2a283bc] 
succeeded in 0.125s:None
[2020-10-09 14:12:05,838: INFO/MainProcess] Received task: 
send_requests.tasks.update_global_list[d382a795-8041-4331-a9ca-d4d74b2c8982]
[2020-10-09 14:12:05,840: WARNING/SpawnPoolWorker-1] my_global_var : 
[['first'],['first']]
[2020-10-09 14:12:05,842: INFO/SpawnPoolWorker-1] Task 
send_requests.tasks.update_global_list[d382a795-8041-4331-a9ca-d4d74b2c8982] 
succeeded in 0.015999999945051968s: None

看起来不同的工作进程彼此之间并不同步。我很困惑。如何同步它们？

解决方法

在分布式环境中使用全局变量只会引起麻烦……如果您使用单个Celery worker和线程作为并发类型，则可能有效。 -典型的解决方案是使用缓存服务器（Redis，memcached或类似服务器）。

为什么？-所有工作进程将具有自己的my_global_var版本，因此当运行向其添加内容的任务时，它将在那中修改my_global_var >工作进程...

版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 dio@foxmail.com 举报，一经查实，本站将立刻删除。

相关推荐

matplotlib报错：AttributeError: module 'backend_interagg' has no attribute 'FigureCanvas'. Did you mean: 'FigureCanvasAgg'?
使用本地python环境可以成功执行 import pandas as pd import matplotlib.pyplot as plt # 设置字体 plt.rcParams[&#39;font.sans-serif&#39;] = [&#39;SimHei&#39;] # 能正确显示负号 p
gitlab登录失败，报错：This challenge page was accidentally cached by an intermediary and is no longer available.
设置时间 控制面板
后端开发常见错误
错误1：Request method ‘DELETE‘ not supported 错误还原：controller层有一个接口，访问该接口时报错：Request method ‘DELETE‘ not supported 错误原因：没有接收到前端传入的参数，修改为如下 参考 错误2：cannot r
docker常见错误
错误1：启动docker镜像时报错：Error response from daemon: driver failed programming external connectivity on endpoint quirky_allen 解决方法：重启docker -&gt; systemctl r
idea常见错误
错误1：private field ‘xxx‘ is never assigned 按Altʾnter快捷键，选择第2项 参考：https://blog.csdn.net/shi_hong_fei_hei/article/details/88814070 错误2：启动时报错，不能找到主启动类 #
pip安装依赖失败
报错如下，通过源不能下载，最后警告pip需升级版本 Requirement already satisfied: pip in c:\users\ychen\appdata\local\programs\python\python310\lib\site-packages (22.0.4) Coll
maven常见错误
错误1：maven打包报错 错误还原：使用maven打包项目时报错如下 [ERROR] Failed to execute goal org.apache.maven.plugins:maven-resources-plugin:3.2.0:resources (default-resources)
cloud项目中常见错误
错误1：服务调用时报错 服务消费者模块assess通过openFeign调用服务提供者模块hires 如下为服务提供者模块hires的控制层接口 @RestController @RequestMapping(&quot;/hires&quot;) public class FeignControl
springboot常见错误
错误1：运行项目后报如下错误 解决方案 报错2：Failed to execute goal org.apache.maven.plugins:maven-compiler-plugin:3.8.1:compile (default-compile) on project sb 解决方案：在pom.
springmvc拦截器中使用redisTemplate报空指针异常
参考 错误原因 过滤器或拦截器在生效时，redisTemplate还没有注入 解决方案：在注入容器时就生效 @Component //项目运行时就注入Spring容器 public class RedisBean { @Resource private RedisTemplate&lt;String
vite报错：npm ERR! command C:WINDOWSsystem32cmd.exe /d /s /c C:UsersychenAppDataLocalTempnpx-dde1c848.cmd
使用vite构建项目报错 C:\Users\ychen\work&gt;npm init @vitejs/app @vitejs/create-app is deprecated, use npm init vite instead C:\Users\ychen\AppData\Local\npm-
VM VirtualBox中安装centos时报错：设置基础软件仓库时出错
参考1 参考2 解决方案 # 点击安装源 协议选择 http:// 路径填写 mirrors.aliyun.com/centos/8.3.2011/BaseOS/x86_64/os URL类型 软件库URL 其他路径 # 版本 7 mirrors.aliyun.com/centos/7/os/x86
kafka常见问题
报错1 [root@slave1 data_mocker]# kafka-console-consumer.sh --bootstrap-server slave1:9092 --topic topic_db [2023-12-19 18:31:12,770] WARN [Consumer clie
hive常见问题
错误1 # 重写数据 hive (edu)&gt; insert overwrite table dwd_trade_cart_add_inc &gt; select data.id, &gt; data.user_id, &gt; data.course_id, &gt; date_format(
装载数据时报错：Failed to execute spark task, with exception 'org.apache.hadoop.hive.ql.metadata.HiveException(Failed to create spark client.)'
错误1 hive (edu)&gt; insert into huanhuan values(1,&#39;haoge&#39;); Query ID = root_20240110071417_fe1517ad-3607-41f4-bdcf-d00b98ac443e Total jobs = 1
flume常见问题
报错1：执行到如下就不执行了，没有显示Successfully registered new MBean. [root@slave1 bin]# /usr/local/software/flume-1.9.0/bin/flume-ng agent -n a1 -c /usr/local/softwa
hadoop集群常见错误
虚拟及没有启动任何服务器查看jps会显示jps，如果没有显示任何东西 [root@slave2 ~]# jps 9647 Jps 解决方案 # 进入/tmp查看 [root@slave1 dfs]# cd /tmp [root@slave1 tmp]# ll 总用量 48 drwxr-xr-x. 2
hive常见错误
报错1 hive&gt; show databases; OK Failed with exception java.io.IOException:java.lang.RuntimeException: Error in configuring object Time taken: 0.474 se
Linux常见错误
报错1 [root@localhost ~]# vim -bash: vim: 未找到命令 安装vim yum -y install vim* # 查看是否安装成功 [root@hadoop01 hadoop]# rpm -qa |grep vim vim-X11-7.4.629-8.el7_9.x
使用sqoop一直卡在：mapreduce.Job: Running job: job_1703173956074_0001
修改hadoop配置 vi /usr/local/software/hadoop-2.9.2/etc/hadoop/yarn-site.xml # 添加如下 &lt;configuration&gt; &lt;property&gt; &lt;name&gt;yarn.nodemanager.res
pythonJavaScriptjavaHTMLPHPreactjsC#AndroidCSSNode.jssqlrpython-3.xMysqLjQueryc++pandasFlutterangularIOSdjangolinuxswifttypescript路由器JSON路由器设置无线路由器h3c华三华三路由器设置华三路由器电脑软件教程arraysdocker软件图文教程Cvue.jslaravelspring-boot