分布式进程

顾名思义：就是把进程的分布不同的机器上执行，进程之间通过网络进行通信。

对比与之前所有的多进程模式，都是只建立再本地的服务进程上。

分布式爬虫的示例

思路：

• 服务端：服务生产需要抓取链接地址

• 客户端（可以多开）：其他机器上的进程负责地址请求处理并返回返回

客户端会从队列自动获取没有被消费处理的消息

服务端：serve.py

#!/usr/bin/evn python
# -*- coding: utf-8 -*-
"""
-------------------------------------------------
   文件名称 :     task_manager
   文件功能描述 :   功能描述
   创建人 :       小钟同学
   创建时间 :          2021/5/8
-------------------------------------------------
   修改描述-2021/5/8:         
-------------------------------------------------
"""
import random, time, queue
from multiprocessing.managers import BaseManager
from multiprocessing import freeze_support
from threading import Thread
# 定义发送任务的队列:
task_queue = queue.Queue()
# 接收结果的队列:
result_queue = queue.Queue()


# 自定义函数re_task_queue
def re_task_queue():
    # global task_queue
    return task_queue


# 自定义函数re_result_queue
def re_result_queue():
    # global result_queue
    return result_queue


# 从BaseManager继承的QueueManager:
class QueueManager(BaseManager):
    pass

def get_wait_result(result):
    # 循环的获取队列处理结果信息
    while True:
        r = result.get(True)  # 阻塞等待结果队列有结果
        print('阻塞等待队列处理结果:', r)

if __name__ == '__main__':

    freeze_support()


    # 把两个Queue都注册到网络上, callable参数关联了Queue对象:
    QueueManager.register('get_task_queue', callable=re_task_queue)
    QueueManager.register('get_result_queue', callable=re_result_queue)

    # 绑定端口5000, 设置验证码'abc':
    manager = QueueManager(address=('127.0.0.1', 8000), authkey=b'abc')
    # 启动Queue:
    manager.start()

    # 获得通过网络访问的Queue对象:
    task = manager.get_task_queue()
    result = manager.get_result_queue()

    t = Thread(target=get_wait_result, args=(result,))
    t.start()
    while True:
        msg = input('请输入需要抓取URL地址：>>>: ').strip()
        if not msg: continue
        # 远程的链接的关闭了话，则会提示异常
        if msg.upper()=='EXIT':
            break
        # 丢一个任务进入
        task.put(msg)

    manager.shutdown()
    print('关闭服务端')

客户端：client.py

#!/usr/bin/evn python
# -*- coding: utf-8 -*-
"""
-------------------------------------------------
   文件名称 :     task_worker
   文件功能描述 :   功能描述
   创建人 :       小钟同学
   创建时间 :          2021/5/8
-------------------------------------------------
   修改描述-2021/5/8:         
-------------------------------------------------
"""
import time, sys, queue
from multiprocessing.managers import BaseManager

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

# 由于这个QueueManager只从网络上获取Queue，所以注册时只提供名字:
QueueManager.register('get_task_queue')
QueueManager.register('get_result_queue')

# 连接到服务器，也就是运行task_master.py的机器:
server_addr = '127.0.0.1'
print('链接服务器：%s...' % server_addr)
# 端口和验证码注意保持与task_master.py设置的完全一致:
m = QueueManager(address=(server_addr, 8000), authkey=b'abc')
# 从网络连接:
m.connect()
# 获取Queue的对象:
task = m.get_task_queue()
result = m.get_result_queue()
# 从task队列取任务,并把结果写入result队列:
import requests
requests.packages.urllib3.disable_warnings()
while True:
    try:
        # n = task.get(timeout=1)
        n = task.get(True)  #阻塞等待结果队列有结果
        print('客户端开始获取任务开始执行：%s'%(n))
        resultmsg = requests.get(n,verify =  False).text
        print('客户端处理结果完成，丢到服务端结果队列里面去了！')
        result.put(resultmsg)
    except Exception:
        print("其他异常")
# 处理结束:
print('客户端处理队列里面所有的任务！！')

首先需要先启动服务端，再启动客户端，不然客户端连接不到客户端会报异常

服务端端输出结果：

请输入需要抓取URL地址：>>>: http://www.baidu.com
请输入需要抓取URL地址：>>>: 阻塞等待队列处理结果: <!DOCTYPE html>
<!/html;xxxxxxxxxxxx内容 <img src=//www.baidu.com/img/gs.gif> </p> </div> </div> </div> </body> </html>

客户端输出内容：

链接服务器：127.0.0.1...
客户端开始获取任务开始执行：http://www.baidu.com
客户端处理结果完成，丢到服务端结果队列里面去了！

多开客户端的情况下，可以获取到队列没有被消费的消息！

资料参考

python 分布式进程：https://www.cnblogs.com/tjp40922/p/12709510.html

个人其他博客地址

简书：https://www.jianshu.com/u/d6960089b087

掘金：https://juejin.cn/user/2963939079225608

小钟同学 | 文 【原创】| QQ：308711822

• 1：本文相关描述主要是个人的认知和见解，如有不当之处，还望各位大佬指正。
• 2：关于文章内容，有部分内容参考自互联网整理，如有链接会声明标注；如没有及时标注备注的链接的，如有侵权请联系，我会立即删除处理哟。