Python的分布式函数的简单介绍

Python高性能分布式执行框架-Ray

这是别人说的，咱也不敢说，咱也不敢问 ! 了解大致的逻辑就好.你只需要知道他超级牛逼,超级方便

安装就是简单的 pip install ray , 需要提醒的就是ray现在只有linux编译版本, windows就别想着用了,为了这,我硬生生把开发环境从windows切到了linux.

首先来看一下最简单的Ray程序是如何编写的。

在Ray里，通过Python注解@ray.remote定义remote函数。使用此注解声明的函数都会自带一个默认的方法remote，通过此方法发起的函数调用都是以提交分布式任务的方式异步执行的，函数的返回值是一个对象id，使用ray.get内置操作可以同步获取该id对应的对象

参考:

高性能分布式执行框架——Ray

取代 Python 多进程！伯克利开源分布式框架 Ray

官方文档

基于python的高性能实时并行机器学习框架之Ray介绍

相关问答

Q1: Python的函数都有哪些

【常见的内置函数】

1、enumerate(iterable,start=0)

是python的内置函数，是枚举、列举的意思，对于一个可迭代的(iterable)/可遍历的对象(如列表、字符串)，enumerate将其组成一个索引序列，利用它可以同时获得索引和值。

2、zip(*iterables,strict=False)

用于将可迭代的对象作为参数，将对象中对应的元素打包成一个个元组，然后返回由这些元组组成的列表。如果各个迭代器的元素个数不一致，则返回列表长度与最短的对象相同，利用*号操作符，可以将元组解压为列表。

3、filter(function,iterable)

filter是将一个序列进行过滤，返回迭代器的对象，去除不满足条件的序列。

4、isinstance(object,classinfo)

是用来判断某一个变量或者是对象是不是属于某种类型的一个函数，如果参数object是classinfo的实例，或者object是classinfo类的子类的一个实例，

返回True。如果object不是一个给定类型的的对象， 则返回结果总是False

5、eval(expression[,globals[,locals]])

用来将字符串str当成有效的表达式来求值并返回计算结果，表达式解析参数expression并作为Python表达式进行求值(从技术上说是一个条件列表)，采用globals和locals字典作为全局和局部命名空间。

【常用的句式】

1、format字符串格式化

format把字符串当成一个模板，通过传入的参数进行格式化，非常实用且强大。

2、连接字符串

常使用+连接两个字符串。

3、if...else条件语句

Python条件语句是通过一条或多条语句的执行结果(True或者False)来决定执行的代码块。其中if...else语句用来执行需要判断的情形。

4、for...in、while循环语句

循环语句就是遍历一个序列，循环去执行某个操作，Python中的循环语句有for和while。

5、import导入其他脚本的功能

有时需要使用另一个python文件中的脚本，这其实很简单，就像使用import关键字导入任何模块一样。

Q2: [Django] celery的替代品 funboost

Django开发web应用的过程中，一个老大难问题是异步调度问题。例如用户传来一个非常耗时的请求，这时候最好的处理方式是先把这个操作请求记录下来，先响应请求，等后面有空的时候再去计算，而不是让用户干等着着急。

这种优化方式就是典型的生产者+消息队列+消费者设计模式，而Django框架本身并没有直接提供该设计模式的实现，大多教程都是利用第三方组件celery+redis来实现这个调度。

遗憾的是celery和redis官方都不支持windows，而我习惯的开发环境还是win10，所以需要找一个替代品。经过调研，发现了一个很好的【python分布式函数调度框架——funboost】. 它的优点很多，对Django开发来说，最大的亮点是完全无需启动第三方服务，即可实现生产消费设计模式。一个 pip install funboost 即可干活，开箱即用。它可以使用SQLite文件来做消息队列，足以应对小型应用开发。当然也可以使用Kafka这种高级的消息中间件，实现高可用。

要说缺点吧，这个组件的日志打印太啰嗦，而且没有提供关闭选项，控制台已被它刷屏。

Q3: 如何用 Python 构建一个简单的分布式系统

分布式爬虫概览

何谓分布式爬虫？

通俗的讲，分布式爬虫就是多台机器多个

spider

对多个

url

的同时处理问题，分布式的方式可以极大提高程序的抓取效率。

构建分布式爬虫通畅需要考虑的问题

（1）如何能保证多台机器同时抓取同一个URL？

（2）如果某个节点挂掉，会不会影响其它节点，任务如何继续？

（3）既然是分布式，如何保证架构的可伸缩性和可扩展性？不同优先级的抓取任务如何进行资源分配和调度？

基于上述问题，我选择使用celery作为分布式任务调度工具，是分布式爬虫中任务和资源调度的核心模块。它会把所有任务都通过消息队列发送给各个分布式节点进行执行，所以可以很好的保证url不会被重复抓取；它在检测到worker挂掉的情况下，会尝试向其他的worker重新发送这个任务信息，这样第二个问题也可以得到解决；celery自带任务路由，我们可以根据实际情况在不同的节点上运行不同的抓取任务（在实战篇我会讲到）。本文主要就是带大家了解一下celery的方方面面(有celery相关经验的同学和大牛可以直接跳过了)

Celery知识储备

celery基础讲解

按celery官网的介绍来说

Celery

是一个简单、灵活且可靠的，处理大量消息的分布式系统，并且提供维护这样一个系统的必需工具。它是一个专注于实时处理的任务队列，同时也支持任务调度。

下面几个关于celery的核心知识点

broker：翻译过来叫做中间人。它是一个消息传输的中间件，可以理解为一个邮箱。每当应用程序调用celery的异步任务的时候，会向broker传递消息，而后celery的worker将会取到消息，执行相应程序。这其实就是消费者和生产者之间的桥梁。

backend:

通常程序发送的消息，发完就完了，可能都不知道对方时候接受了。为此，celery实现了一个backend，用于存储这些消息以及celery执行的一些消息和结果。

worker:

Celery类的实例，作用就是执行各种任务。注意在celery3.1.25后windows是不支持celery

worker的！

producer:

发送任务，将其传递给broker

beat:

celery实现的定时任务。可以将其理解为一个producer，因为它也是通过网络调用定时将任务发送给worker执行。注意在windows上celery是不支持定时任务的！

下面是关于celery的架构示意图，结合上面文字的话应该会更好理解

由于celery只是任务队列，而不是真正意义上的消息队列，它自身不具有存储数据的功能，所以broker和backend需要通过第三方工具来存储信息，celery官方推荐的是

RabbitMQ和Redis，另外mongodb等也可以作为broker或者backend，可能不会很稳定，我们这里选择Redis作为broker兼backend。

实际例子

先安装celery

pip

install

celery

我们以官网给出的例子来做说明，并对其进行扩展。首先在项目根目录下，这里我新建一个项目叫做celerystudy，然后切换到该项目目录下，新建文件tasks.py，然后在其中输入下面代码

这里我详细讲一下代码：我们先通过app=Celery()来实例化一个celery对象，在这个过程中，我们指定了它的broker，是redis的db

2,也指定了它的backend,是redis的db3,

broker和backend的连接形式大概是这样

redis://:password@hostname:port/db_number

然后定义了一个add函数，重点是@app.task，它的作用在我看来就是将add()

注册为一个类似服务的东西，本来只能通过本地调用的函数被它装饰后，就可以通过网络来调用。这个tasks.py中的app就是一个worker。它可以有很多任务，比如这里的任务函数add。我们再通过在命令行切换到项目根目录，执行

celery

-A

tasks

worker

-l

info

启动成功后就是下图所示的样子

这里我说一下各个参数的意思，-A指定的是app(即Celery实例)所在的文件模块，我们的app是放在tasks.py中，所以这里是

tasks；worker表示当前以worker的方式运行，难道还有别的方式？对的，比如运行定时任务就不用指定worker这个关键字;

-l

info表示该worker节点的日志等级是info，更多关于启动worker的参数(比如-c、-Q等常用的)请使用

celery

worker

--help

进行查看

将worker启动起来后，我们就可以通过网络来调用add函数了。我们在后面的分布式爬虫构建中也是采用这种方式分发和消费url的。在命令行先切换到项目根目录，然后打开python交互端

from

tasks

import

addrs

=

add.delay(2,

2)

这里的add.delay就是通过网络调用将任务发送给add所在的worker执行，这个时候我们可以在worker的界面看到接收的任务和计算的结果。

这里是异步调用，如果我们需要返回的结果，那么要等rs的ready状态true才行。这里add看不出效果，不过试想一下，如果我们是调用的比较占时间的io任务，那么异步任务就比较有价值了

上面讲的是从Python交互终端中调用add函数，如果我们要从另外一个py文件调用呢？除了通过import然后add.delay()这种方式，我们还可以通过send_task()这种方式，我们在项目根目录另外新建一个py文件叫做

excute_tasks.py，在其中写下如下的代码

from

tasks

import

addif

__name__

==

'__main__':

add.delay(5,

10)

这时候可以在celery的worker界面看到执行的结果

此外，我们还可以通过send_task()来调用，将excute_tasks.py改成这样

这种方式也是可以的。send_task()还可能接收到为注册（即通过@app.task装饰）的任务，这个时候worker会忽略这个消息

定时任务

上面部分讲了怎么启动worker和调用worker的相关函数，这里再讲一下celery的定时任务。

爬虫由于其特殊性，可能需要定时做增量抓取，也可能需要定时做模拟登陆，以防止cookie过期，而celery恰恰就实现了定时任务的功能。在上述基础上，我们将tasks.py文件改成如下内容

然后先通过ctrl+c停掉前一个worker，因为我们代码改了，需要重启worker才会生效。我们再次以celery

-A

tasks

worker

-l

info这个命令开启worker。

这个时候我们只是开启了worker，如果要让worker执行任务，那么还需要通过beat给它定时发送，我们再开一个命令行，切换到项目根目录，通过

这样就表示定时任务已经开始运行了。

眼尖的同学可能看到我这里celery的版本是3.1.25，这是因为celery支持的windows最高版本是3.1.25。由于我的分布式微博爬虫的worker也同时部署在了windows上，所以我选择了使用

3.1.25。如果全是linux系统，建议使用celery4。

此外，还有一点需要注意，在celery4后，定时任务（通过schedule调度的会这样，通过crontab调度的会马上执行）会在当前时间再过定时间隔执行第一次任务，比如我这里设置的是60秒的间隔，那么第一次执行add会在我们通过celery

beat

-A

tasks

-l

info启动定时任务后60秒才执行；celery3.1.25则会马上执行该任务

Q4: Python中的多进程与多线程/分布式该如何使用

Python提供了非常好用的多进程包multiprocessing，你只需要定义一个函数，Python会替你完成其他所有事情。

借助这个包，可以轻松完成从单进程到并发执行的转换。

1、新建单一进程

如果我们新建少量进程，可以如下：

import multiprocessing

import time

def func(msg):

for i in xrange(3):

print msg

time.sleep(1)

if __name__ == "__main__":

p = multiprocessing.Process(target=func, args=("hello", ))

p.start()

p.join()

print "Sub-process done."12345678910111213

2、使用进程池

是的，你没有看错，不是线程池。它可以让你跑满多核CPU，而且使用方法非常简单。

注意要用apply_async，如果落下async，就变成阻塞版本了。

processes=4是最多并发进程数量。

import multiprocessing

import time

def func(msg):

for i in xrange(3):

print msg

time.sleep(1)

if __name__ == "__main__":

pool = multiprocessing.Pool(processes=4)

for i in xrange(10):

msg = "hello %d" %(i)

pool.apply_async(func, (msg, ))

pool.close()

pool.join()

print "Sub-process(es) done."12345678910111213141516

3、使用Pool，并需要关注结果

更多的时候，我们不仅需要多进程执行，还需要关注每个进程的执行结果，如下：

import multiprocessing

import time

def func(msg):

for i in xrange(3):

print msg

time.sleep(1)

return "done " + msg

if __name__ == "__main__":

pool = multiprocessing.Pool(processes=4)

result = []

for i in xrange(10):

msg = "hello %d" %(i)

result.append(pool.apply_async(func, (msg, )))

pool.close()

pool.join()

for res in result:

print res.get()

print "Sub-process(es) done."1234567891011121314151617181920

2014.12.25更新

根据网友评论中的反馈，在Windows下运行有可能崩溃(开启了一大堆新窗口、进程)，可以通过如下调用来解决：

multiprocessing.freeze_support()1

附录（自己的脚本）：

#!/usr/bin/python

import threading

import subprocess

import datetime

import multiprocessing

def dd_test(round, th):

test_file_arg = 'of=/zbkc/test_mds_crash/1m_%s_%s_{}' %(round, th)

command = "seq 100 | xargs -i dd if=/dev/zero %s bs=1M count=1" %test_file_arg

print command

subprocess.call(command,shell=True,stdout=open('/dev/null','w'),stderr=subprocess.STDOUT)

def mds_stat(round):

p = subprocess.Popen("zbkc mds stat", shell = True, stdout = subprocess.PIPE)

out = p.stdout.readlines()

if out[0].find('active') != -1:

command = "echo '0205pm %s round mds status OK, %s' /round_record" %(round, datetime.datetime.now())

command_2 = "time (ls /zbkc/test_mds_crash/) 2/round_record"

command_3 = "ls /zbkc/test_mds_crash | wc -l /round_record"

subprocess.call(command,shell=True)

subprocess.call(command_2,shell=True)

subprocess.call(command_3,shell=True)

return 1

else:

command = "echo '0205 %s round mds status abnormal, %s, %s' /round_record" %(round, out[0], datetime.datetime.now())

subprocess.call(command,shell=True)

return 0

#threads = []

for round in range(1, 1600):

pool = multiprocessing.Pool(processes = 10) #使用进程池

for th in range(10):

# th_name = "thread-" + str(th)

# threads.append(th_name) #添加线程到线程列表

# threading.Thread(target = dd_test, args = (round, th), name = th_name).start() #创建多线程任务

pool.apply_async(dd_test, (round, th))

pool.close()

pool.join()

#等待线程完成

# for t in threads:

# t.join()

if mds_stat(round) == 0:

subprocess.call("zbkc -s",shell=True)

break

Q5: Python分布式进程中你会遇到的坑

写在前面

小惊大怪

你是不是在用Python3或者在windows系统上编程？最重要的是你对进程和线程不是很清楚？那么恭喜你，在python分布式进程中，会有坑等着你去挖。。。(hahahaha，此处允许我吓唬一下你)开玩笑的啦，不过，如果你知道序列中不支持匿名函数，那这个坑就和你say byebye了。好了话不多数，直接进入正题。

分布式进程

正如大家所知道的Process比Thread更稳定，而且Process可以分布到多台机器上，而Thread最多只能分布到同一台机器的多个CPU上。Python的multiprocessing模块不但支持多进程，其中managers子模块还支持把多进程分布到多台机器上。一个服务进程可以作为调度者，将任务分布到其他多个进程中，依靠网络通信。由于managers模块封装很好，不必了解网络通信的细节，就可以很容易地编写分布式多进程程序。

代码记录

举个例子

如果我们已经有一个通过Queue通信的多进程程序在同一台机器上运行，现在，由于处理任务的进程任务繁重，希望把发送任务的进程和处理任务的进程分布到两台机器上，这应该怎么用分布式进程来实现呢？你已经知道了原有的Queue可以继续使用，而且通过managers模块把Queue通过网络暴露出去，就可以让其他机器的进程来访问Queue了。好，那我们就这么干！

写个task_master.py

我们先看服务进程。服务进程负责启动Queue，把Queue注册到网络上，然后往Queue里面写入任务。

请注意，当我们在一台机器上写多进程程序时，创建的Queue可以直接拿来用，但是，在分布式多进程环境下，添加任务到Queue不可以直接对原始的task_queue进行操作，那样就绕过了QueueManager的封装，必须通过manager.get_task_queue()获得的Queue接口添加。然后，在另一台机器上启动任务进程（本机上启动也可以）

写个task_worker.py

任务进程要通过网络连接到服务进程，所以要指定服务进程的IP。

运行结果

现在，可以试试分布式进程的工作效果了。先启动task_master.py服务进程：

task_master.py进程发送完任务后，开始等待result队列的结果。现在启动task_worker.py进程：

看到没，结果都出错了，我们好好分析一下到底哪出错了。。。

错误分析

在task_master.py的报错提示中，我们知道它说lambda错误，这是因为序列化不支持匿名函数，所以我们得修改代码，重新对queue用QueueManager进行封装放到网络中。

其中task_queue和result_queue是两个队列，分别存放任务和结果。它们用来进行进程间通信，交换对象。

因为是分布式的环境，放入queue中的数据需要等待Workers机器运算处理后再进行读取，这样就需要对queue用QueueManager进行封装放到网络中，这是通过上面的2行代码来实现的。我们给return_task_queue的网络调用接口取了一个名get_task_queue,而return_result_queue的名字是get_result_queue，方便区分对哪个queue进行操作。task.put(n)即是对task_queue进行写入数据，相当于分配任务。而result.get()即是等待workers机器处理后返回的结果。

值得注意 在windows系统中你必须要写IP地址，而其他操作系统比如linux操作系统则就不要了。

修改后的代码

在task_master.py中修改如下：

在task_worker.py中修改如下：

先运行task_master.py,然后再运行task_worker.py

(1)task_master.py运行结果如下

(2)task_worker.py运行结果如下

知识补充

这个简单的Master/Worker模型有什么用？其实这就是一个简单但真正的分布式计算，把代码稍加改造，启动多个worker，就可以把任务分布到几台甚至几十台机器上，比如把计算n*n的代码换成发送邮件，就实现了邮件队列的异步发送。

Queue对象存储在哪？注意到task_worker.py中根本没有创建Queue的代码，所以，Queue对象存储在task_master.py进程中：

而Queue之所以能通过网络访问，就是通过QueueManager实现的。由于QueueManager管理的不止一个Queue，所以，要给每个Queue的网络调用接口起个名字，比如get_task_queue。task_worker这里的QueueManager注册的名字必须和task_manager中的一样。对比上面的例子，可以看出Queue对象从另一个进程通过网络传递了过来。只不过这里的传递和网络通信由QueueManager完成。

authkey有什么用？这是为了保证两台机器正常通信，不被其他机器恶意干扰。如果task_worker.py的authkey和task_master.py的authkey不一致，肯定连接不上。