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

Irene181
• 阅读 1583

一、前言

在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。

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

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

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

三、总结

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



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

**-----**------**-----**---**** End **-----**--------**-----**-****

往期精彩文章推荐:

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

欢迎各位大佬点击链接加入群聊【helloworld开发者社区】:https://jq.qq.com/?_wv=1027&k=mBlk6nzX进群交流IT技术热点。

本文转自 https://mp.weixin.qq.com/s/C50Xkv_XztqLBOhWIka0cA,如有侵权,请联系删除。

点赞
收藏
评论区
推荐文章
blmius blmius
3年前
MySQL:[Err] 1292 - Incorrect datetime value: ‘0000-00-00 00:00:00‘ for column ‘CREATE_TIME‘ at row 1
文章目录问题用navicat导入数据时,报错:原因这是因为当前的MySQL不支持datetime为0的情况。解决修改sql\mode:sql\mode:SQLMode定义了MySQL应支持的SQL语法、数据校验等,这样可以更容易地在不同的环境中使用MySQL。全局s
皕杰报表之UUID
​在我们用皕杰报表工具设计填报报表时,如何在新增行里自动增加id呢?能新增整数排序id吗?目前可以在新增行里自动增加id,但只能用uuid函数增加UUID编码,不能新增整数排序id。uuid函数说明:获取一个UUID,可以在填报表中用来创建数据ID语法:uuid()或uuid(sep)参数说明:sep布尔值,生成的uuid中是否包含分隔符'',缺省为
Jacquelyn38 Jacquelyn38
3年前
2020年前端实用代码段,为你的工作保驾护航
有空的时候,自己总结了几个代码段,在开发中也经常使用,谢谢。1、使用解构获取json数据let jsonData  id: 1,status: "OK",data: 'a', 'b';let  id, status, data: number   jsonData;console.log(id, status, number )
Stella981 Stella981
3年前
Python3:sqlalchemy对mysql数据库操作,非sql语句
Python3:sqlalchemy对mysql数据库操作,非sql语句python3authorlizmdatetime2018020110:00:00coding:utf8'''
Stella981 Stella981
3年前
Python之time模块的时间戳、时间字符串格式化与转换
Python处理时间和时间戳的内置模块就有time,和datetime两个,本文先说time模块。关于时间戳的几个概念时间戳,根据1970年1月1日00:00:00开始按秒计算的偏移量。时间元组(struct_time),包含9个元素。 time.struct_time(tm_y
Easter79 Easter79
3年前
Twitter的分布式自增ID算法snowflake (Java版)
概述分布式系统中,有一些需要使用全局唯一ID的场景,这种时候为了防止ID冲突可以使用36位的UUID,但是UUID有一些缺点,首先他相对比较长,另外UUID一般是无序的。有些时候我们希望能使用一种简单一些的ID,并且希望ID能够按照时间有序生成。而twitter的snowflake解决了这种需求,最初Twitter把存储系统从MySQL迁移
Wesley13 Wesley13
3年前
00:Java简单了解
浅谈Java之概述Java是SUN(StanfordUniversityNetwork),斯坦福大学网络公司)1995年推出的一门高级编程语言。Java是一种面向Internet的编程语言。随着Java技术在web方面的不断成熟,已经成为Web应用程序的首选开发语言。Java是简单易学,完全面向对象,安全可靠,与平台无关的编程语言。
Stella981 Stella981
3年前
Python process (进程)
进程(process)进程是对各种资源管理的集合,包含对各种资源的调用、内存的管理、网络接口的调用进程要操作CPU必须先启动一个线程,启动一个进程的时候会自动创建一个线程,进程里的第一个线程就是主线程程序执行的实例有唯一的进程标识符(pid)multiprossing模块
Wesley13 Wesley13
3年前
MySQL部分从库上面因为大量的临时表tmp_table造成慢查询
背景描述Time:20190124T00:08:14.70572408:00User@Host:@Id:Schema:sentrymetaLast_errno:0Killed:0Query_time:0.315758Lock_
Python进阶者 Python进阶者
9个月前
Excel中这日期老是出来00:00:00,怎么用Pandas把这个去除
大家好,我是皮皮。一、前言前几天在Python白银交流群【上海新年人】问了一个Pandas数据筛选的问题。问题如下:这日期老是出来00:00:00,怎么把这个去除。二、实现过程后来【论草莓如何成为冻干莓】给了一个思路和代码如下:pd.toexcel之前把这