Python分布式进程

龙哥

发布于 2018-10-22 14:27:28

9280

发布于 2018-10-22 14:27:28

说明：本文是基于Py2.X环境，

分布式进程：

分布式进程是指的是将Process进程分布到多台机器上，充分利用多台机器的性能完成复杂的任务。在Thread和Process中，应当优选Process，因为Process更稳定，而且，Process可以分布到多台机器上，而Thread最多只能分布到同一台机器的多个CPU上。

Python的multiprocessing模块不但支持多进程，其中managers子模块还支持把多进程分布到多台机器上。一个服务进程可以作为调度者，将任务分布到其他多个进程中，依靠网络通信。由于managers模块封装很好，不必了解网络通信的细节，就可以很容易地编写分布式多进程程序。

举个例子：做爬虫程序时，常常会遇到这样的场景，我们想抓取图片的链接地址，将链接地址存放到Queue中，另外的进程负责从Queue中读取链接地址进行下载和存储到本地。现在把这个过程做成分布式，一台机器上的进程负责抓取链接，其它机器上的进程负责下载存储，那么遇到的主要问题是将Queue暴露到网络中，让其它机器进程都可以访问，分布式进程就是将这一个过程进行了封装，我们可以将这个过程称为本队列的网络化。

创建分布式进程需要一个服务进程与任务进程：

服务进程创建：

建立队列Queue，用来进行进程间的通信。服务进程创建任务队列taskqueue,用来作为传递任务给任务进程的通道;服务进程创建结果队列resultqueue，作为任务进程完成任务后回复服务进程的通道。在分布式多进程环境下，必须通过由Queuemanager获得Queue接口来添加任务.
把第一步中建立的队列在网络上注册，暴露给其它进程(主机)，注册后获得网络队列，相当于本队队列的映像.
建立一个险象(Queuemanager（BaseManager）)实例manager,绑定端口和验证口令。
启动第三步中建立的实例，即启动管理manager,监管信息通道
通过管理实例的方法获得通过网络访问的Queue对象，即再把网络队列实体化成可以使用的本地队列.
创建任务到"本地"队列中，自动上传任务到网络队列中，分配给任务进程进行处理。

注意：我这里是基于window操作系统的，linux系统会有所不同

# coding:utf-8
# taskManager.py for win

import Queue
from multiprocessing.managers import BaseManager
from multiprocessing import freeze_support

# 任务个数
task_num = 10

# 定义收发队列
task_queue = Queue.Queue(task_num)
result_queue = Queue.Queue(task_num)


def get_task():
    return task_queue


def get_result():
    return result_queue


# 创建类似的QueueManager

class QueueManager(BaseManager):
    pass


def win_run():
    # windows下绑定调用接口不能使用lambda,所以只能先定义函数再绑定
    QueueManager.register('get_task_queue', callable=get_task)
    QueueManager.register('get_result_queue', callable=get_result)
    # 绑定端口并设置验证口令，windows下需要填写IP地址,Linux下不填，默认为本地
    manager = QueueManager(address=('127.0.0.1', 4000), authkey='qty')

    # 启动
    manager.start()

    # 通过网络获取任务队列和结果队列
    task = manager.get_task_queue()
    result = manager.get_result_queue()

    try:

        # 添加任务
        for i in range(10):
            print 'put task %s...' % i
            task.put(i)
        print 'try get result...'

        for i in range(10):
            print 'result is %s' % result.get(timeout=10)



    except:
        print 'manage error'
    finally:
        # 一定要关闭，否则会报管理未关闭的错误
        manager.shutdown()
        print 'master exit!'


if __name__ == '__main__':
    # windows下多进程可能会出现问题，添加这句可以缓解
    freeze_support()
    win_run()

任务进程创建

使用QueueManager 注册用于获取Queue的方法名称，任务进程只能通过名称来在网络上获取Queue
连接服务器中，端口和验证口令注意保持与服务进程中完全一致
从网络上获取Queue，进行本地化
从Task队列获取任务，并把结果result队列

# coding:utf-8
import time
from multiprocessing.managers import BaseManager


# 创建类似的QueueManager:
class QueueManager(BaseManager):
    pass


# 第一步：使用QueueManager注册用于获取Queue的方法名称
QueueManager.register('get_task_queue')
QueueManager.register('get_result_queue')

# 第二步：连接服务器
server_addr = '127.0.0.1'
print "Connect to server %s" % server_addr

# 端口和验证口令注意保持与服务进程完全一致
m = QueueManager(address=(server_addr, 4000), authkey='qty')

# 从网络连接
m.connect()

# 第三步：获取Queue的对象
task = m.get_task_queue()
result = m.get_result_queue()

# 第四步:从task队列获取任务，并把结果写入result队列:

while not task.empty():
    index = task.get(True, timeout=10)
    print 'run task download %s' % str(index)
    result.put('%s---->success ' % str(index))

# 处理结束
print 'worker exit.'

执行结果

先运行:服务进程得到结果

put task 0...
put task 1...
put task 2...
put task 3...
put task 4...
put task 5...
put task 6...
put task 7...
put task 8...
put task 9...
try get result...

再立即运行：任务进程得到结果，防止进程走完后得不到结果，这里一定要立即执行

Connect to server 127.0.0.1
run task download 0
run task download 1
run task download 2
run task download 3
run task download 4
run task download 5
run task download 6
run task download 7
run task download 8
run task download 9
worker exit.

最后再回头看服务进程窗口的结果

put task 0...
put task 1...
put task 2...
put task 3...
put task 4...
put task 5...
put task 6...
put task 7...
put task 8...
put task 9...
try get result...
result is 0---->success 
result is 1---->success 
result is 2---->success 
result is 3---->success 
result is 4---->success 
result is 5---->success 
result is 6---->success 
result is 7---->success 
result is 8---->success 
result is 9---->success 
master exit!

这就是一个简单但真正的分布式计算，把代码稍加改造，启动多个worker，就把任务分布到几台甚至几十台机器上，实现大规模的分布式爬虫

本文参与腾讯云自媒体分享计划，分享自微信公众号。

原始发表：2017-12-21，如有侵权请联系 cloudcommunity@tencent.com 删除

python