一篇文章带你了解Python的分布式进程接口

小鱼儿 2022-10-15 05:48 307阅读 0赞

点击上方“**Go语言进阶学习**”，进行关注

回复“**Go语言**”即可获赠从入门到进阶共10本电子书

今

日

鸡

汤

离离原上草，一岁一枯荣。

**一、前言**

在Thread和Process中，应当优选Process，因为Process更稳定，而且，Process可以分布到多台机器上，而Thread最多只能分布到同一台机器的多个CPU上。

Python的multiprocessing模块不但支持多进程，其中managers子模块还支持把多进程分布到多台机器上。可以写一个服务进程作为调度者，将任务分布到其他多个进程中，依靠网络通信进行管理。

**二、案例分析**

在做爬虫程序时，抓取某个网站的所有图片，如果使用多进程的话，一般是一个进程负责抓取图片的链接地址，将链接地址放到queue中，另外的进程负责 从queue中取链接地址进行下载和存储到本地。

怎么用分布式进程实现？

一台机器上的进程负责抓取链接地址，其他机器上的进程负责系在存储。那么遇到的主要问题是将queue 暴露到网络中，让其他机器进程都可以访问，分布式进程就是将这个过程进行了封装，可以将这个过程称为本地队列的网络化。

例：

1.py

from multiprocessing.managers import BaseManager
    from multiprocessing import freeze_support, Queue
    # 任务个数
    task_number = 10
    
    
    # 收发队列
    task_quue = Queue(task_number)
    result_queue = Queue(task_number)
    
    
    def get_task():
        return task_quue
    
    
    def get_result():
        return result_queue
    # 创建类似的queueManager
    class QueueManager(BaseManager):
        pass
    
    
    def win_run():
        # 注册在网络上，callable 关联了Queue 对象
        # 将Queue对象在网络中暴露
        # window下绑定调用接口不能直接使用lambda，所以只能先定义函数再绑定
        QueueManager.register('get_task_queue', callable=get_task)
        QueueManager.register('get_result_queue', callable=get_result)
        # 绑定端口和设置验证口令
        manager = QueueManager(address=('127.0.0.1', 8001), authkey='qiye'.encode())
        # 启动管理，监听信息通道
        manager.start()
    
    
        try:
            # 通过网络获取任务队列和结果队列
            task = manager.get_task_queue()
            result = manager.get_result_queue()
    
    
            # 添加任务
            for url in ["ImageUrl_" + str(i) for i in range(10)]:
                print('url is %s' % url)
                task.put(url)
                
            print('try get result')
            for i in range(10):
                print('result is %s' % result.get(timeout=10))
    
    
        except:
            print('Manager error')
        finally:
            manager.shutdown()
    
    
    if __name__ == '__main__':
        freeze_support()
        win_run()

连接服务器，端口和验证口令注意保持与服务器进程中完全一致从网络获取Queue，进行本地化，从task队列获取任务，并且把结果写入result队列

2.py

#coding:utf-8
    import time
    from multiprocessing.managers import BaseManager
    # 创建类似的Manager:
    class Manager(BaseManager):
        pass
    #使用QueueManager注册获取Queue的方法名称
    Manager.register('get_task_queue')
    Manager.register('get_result_queue')
    #连接到服务器:
    server_addr = '127.0.0.1'
    print('Connect to server %s...' % server_addr)
    # 端口和验证口令注意保持与服务进程设置的完全一致:
    m = Manager(address=(server_addr, 8001), authkey='qiye')
    # 从网络连接:
    m.connect()
    #获取Queue的对象:
    task = m.get_task_queue()
    result = m.get_result_queue()
    #从task队列取任务,并把结果写入result队列:
    while(not task.empty()):
            image_url = task.get(True,timeout=5)
            print('run task download %s...' % image_url)
            time.sleep(1)
            result.put('%s--->success'%image_url)
    #结束:
    print('worker exit.')

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

运行结果如下：

获取图片地址,将地址传到2.py。

![cecbcf638e16eb41df4ada472a47dc64.png][]

接收1.py传递的地址，进行图片的下载，控制台显示爬取结果。

![c26a4a1c90238b1dafbf48d80a85508e.png][]

**三、总结**

本文基于Python基础，Python的分布式进程接口简单，封装良好，适合需要把繁重任务分布到多台机器的环境下。通过讲解Queue的作用是用来传递任务和接收结果。

欢迎大家积极尝试，有时候看到别人实现起来很简单，但是到自己动手实现的时候，总会有各种各样的问题，切勿眼高手低，勤动手，才可以理解的更加深刻。

**---\--\---\---\---\--\---** End **---\--\---\--\---\---\--\-**

往期精彩文章推荐：

*  [从Go语言角度剖析关于计算机位的问题][Go]
 *  [一篇文章带你搞定Go语言基础之文件操作][Go 1]
 *  [手把手教你用Go语言打造一款简易TCP端口扫描器][Go_TCP]

![0d4117e84452bda9be3c6b71ebdeb651.png][]

欢迎大家**点赞，****留言，****转发，转载，**感谢大家的相伴与支持

想加入Go学习群请在后台回复【**入群**】

万水千山总是情，点个【**在看**】行不行

[cecbcf638e16eb41df4ada472a47dc64.png]: /images/20221014/fc85e56a32bf43f18c73c2606d308ead.png
[c26a4a1c90238b1dafbf48d80a85508e.png]: /images/20221014/d1aa16dfad9640b2a1338cca21fea4d9.png
[Go]: http://mp.weixin.qq.com/s?__biz=MzA5MjkxNDU2Ng%3D%3D&chksm=90649585a7131c937195be0a1535ec629ee3f78b244bfca17899696119897174b26575a630d0&idx=1&mid=2247484268&scene=21&sn=66569be8a278ac48e5944dd6aaa8aa72#wechat_redirect
[Go 1]: http://mp.weixin.qq.com/s?__biz=MzA5MjkxNDU2Ng%3D%3D&chksm=906495dda7131ccbd27a1b996720c80082d8222fbbaa067e6bdd2a89dfee834711b59e8aeb1b&idx=1&mid=2247484212&scene=21&sn=10c771eadabe2b7f171b8c40ec721c5e#wechat_redirect
[Go_TCP]: http://mp.weixin.qq.com/s?__biz=MzA5MjkxNDU2Ng%3D%3D&chksm=906495c0a7131cd604f2de6cc74b23c5197c050dfc1abe4be7682fcdbadc4a56398b39fc978a&idx=1&mid=2247484201&scene=21&sn=dcf2d1d74275661a7a20bdf45036de96#wechat_redirect
[0d4117e84452bda9be3c6b71ebdeb651.png]: /images/20221014/0aeea1bd4af84a8faf4dcb059b97ffed.png