栏目分类:
子分类:
返回
名师互学网用户登录
快速导航关闭
当前搜索
当前分类
子分类
实用工具
热门搜索
名师互学网 > IT > 面试经验 > 面试问答

具有多处理功能的Celery并行分布式任务

面试问答 更新时间: 发布时间: IT归档 最新发布 模块sitemap 名妆网 法律咨询 聚返吧 英语巴士网 伯小乐 网商动力

具有多处理功能的Celery并行分布式任务

您的目标是:

  1. 将您的工作分配到许多机器(分布式计算/分布式并行处理)
  2. 将工作分配到所有CPU上的给定计算机上(多处理/线程)

芹菜可以很轻松地为您做这两个。首先要了解的是,每个celery
worker默认配置为运行与系统上可用CPU内核数量一样多的任务:

并发是用于同时处理您的任务的前叉工作进程的数量,当所有这些工作都忙于工作时,新任务将必须等待其中一项任务完成才能被处理。

缺省的并发数是该机器(包括核心)上的CPU数 ,您可以使用-
c选项指定自定义数。没有建议值,因为最佳数量取决于许多因素,但是如果您的任务主要是受I /
O约束的,那么您可以尝试增加它,实验表明,很少会增加两倍以上的CPU数量有效,并且有可能降低性能。

这意味着每个任务都不必担心使用多处理/线程来利用多个CPU /内核。相反,芹菜将同时运行足够的任务以使用每个可用的CPU。

有了这一点,下一步就是创建一个任务来处理您的的某些子集

list_of_millions_of_ids
。您这里有两个选择-
一个是让每个任务处理一个ID,因此您要运行N个任务,其中
N ==len(list_of_millions_of_ids)
。这将确保工作在所有任务中平均分配,因为永远不会出现一个工人提前完成而只是在等待的情况。如果需要工作,可以将ID移出队列。您可以使用芹菜来做到这一点(如John
Doe所述)
group

task.py:

@app.taskdef process_id(item):    id = item #long complicated equation here    database.objects(newid=id).save()

并执行任务:

from celery import groupfrom tasks import process_idjobs = group(process_id.s(item) for item in list_of_millions_of_ids)result = jobs.apply_async()

另一种选择是将列表分成较小的部分,然后将这些部分分发给您的工作人员。这种方法冒着浪费一些周期的风险,因为您可能最终会有些工人在等待而其他人仍在工作。但是,芹菜文档指出这种担心通常是没有根据的:

有些人可能会担心将任务分块会导致并行度降低,但是对于繁忙的集群却很少如此,实际上,由于避免了消息传递的开销,这可能会大大提高性能。

因此,由于减少了消息传递开销,您可能会发现对列表进行分块并将分块分配给每个任务的效果更好。您可能还可以通过以下方式来减轻数据库的负担:计算每个ID,将其存储在列表中,然后在完成后将整个列表添加到DB中,而不是一次执行一个ID
。分块方法看起来像这样

task.py:

@app.taskdef process_ids(items):    for item in items:        id = item #long complicated equation here        database.objects(newid=id).save() # Still adding one id at a time, but you don't have to.

并开始任务:

from tasks import process_idsjobs = process_ids.chunks(list_of_millions_of_ids, 30) # break the list into 30 chunks. Experiment with what number works best here.jobs.apply_async()

您可以尝试一下分块大小可以为您带来最佳结果的方法。您希望找到一个最佳的位置,在其中减少消息传递的开销,同时又要保持足够小的大小,以免最终导致工作人员完成工作的速度比另一个工作人员快得多,然后无所事事地等待着。



转载请注明:文章转载自 www.mshxw.com
本文地址:https://www.mshxw.com/it/626469.html
我们一直用心在做
关于我们 文章归档 网站地图 联系我们

版权所有 (c)2021-2022 MSHXW.COM

ICP备案号:晋ICP备2021003244-6号