首页 > python教程

python分布式编程实现过程解析

时间:2021-03-06 python教程 查看: 776

分布式编程的难点在于:

1.服务器之间的通信,主节点如何了解从节点的执行进度,并在从节点之间进行负载均衡和任务调度;

2.如何让多个服务器上的进程访问同一资源的不同部分进行执行

第一部分涉及到网络编程的底层细节

第二个问题让我联想到hdfs的一些功能。

首先分布式进程还是解决的是单机单进程无法处理的大数据量大计算量的问题,希望能加通过一份代码(最多主+从两份)来并行执行一个大任务。

这就面临两个问题,首先将程序分布到多台服务器,其次将输入数据分配给多台服务器。

第一个问题相对比较简单,毕竟程序一般不会太长,即便是超级jar包的spark程序,也不过百兆。

但数据里不同,如今企业级别的数据动辄GB、TB,如果在分布式程序执行之前首先要进行大容量数据的转移,显然是不可取的。

这时候我们就需要一个中央共享数据源,所有服务器都可以对这个数据源进行并行存取(块block),这就已经非常接近hdfs的功能。

因为在hdfs中,集群中的多台服务器共享同一个hdfs,每台机器访问hdfs就像访问本地数据一样(还是稍微慢一点);

计算任务执行完之后,每台服务器还可以将自己的计算结果写回hdfs,每台服务器的结果被存储成了结果目录中的小文件。

# task_master.py

import random, time, queue
from multiprocessing.managers import BaseManager

# 发送任务的队列:
task_queue = queue.Queue()
# 接收结果的队列:
result_queue = queue.Queue()

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

# 把两个Queue都注册到网络上, callable参数关联了Queue对象:
QueueManager.register('get_task_queue', callable=lambda: task_queue)
QueueManager.register('get_result_queue', callable=lambda: result_queue)
# 绑定端口5000, 设置验证码'abc':
manager = QueueManager(address=('', 5000), authkey=b'abc')
# 启动Queue:
manager.start()
# 获得通过网络访问的Queue对象:
task = manager.get_task_queue()
result = manager.get_result_queue()
# 放几个任务进去:
for i in range(10):
  n = random.randint(0, 10000)
  print('Put task %d...' % n)
  task.put(n)
# 从result队列读取结果:
print('Try get results...')
for i in range(10):
  r = result.get(timeout=10)
  print('Result: %s' % r)
# 关闭:
manager.shutdown()
print('master exit.')
# task_worker.py

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('Connect to server %s...' % server_addr)
# 端口和验证码注意保持与task_master.py设置的完全一致:
m = QueueManager(address=(server_addr, 5000), authkey=b'abc')
# 从网络连接:
m.connect()
# 获取Queue的对象:
task = m.get_task_queue()
result = m.get_result_queue()
# 从task队列取任务,并把结果写入result队列:
for i in range(10):
  try:
    n = task.get(timeout=1)
    print('run task %d * %d...' % (n, n))
    r = '%d * %d = %d' % (n, n, n*n)
    time.sleep(1)
    result.put(r)
  except Queue.Empty:
    print('task queue is empty.')
# 处理结束:
print('worker exit.')

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持python博客。

展开全文
上一篇:python manage.py runserver流程解析
下一篇:Python 调用C++封装的进一步探索交流
输入字:
相关知识
Python 实现图片色彩转换案例

我们在看动漫、影视作品中,当人物在回忆过程中,体现出来的画面一般都是黑白或者褐色的。本文将提供将图片色彩转为黑白或者褐色风格的案例详解,感兴趣的小伙伴可以了解一下。

python初学定义函数

这篇文章主要为大家介绍了python的定义函数,具有一定的参考价值,感兴趣的小伙伴们可以参考一下,希望能够给你带来帮助,希望能够给你带来帮助

图文详解Python如何导入自己编写的py文件

有时候自己写了一个py文件,想要把它导入到另一个py文件里面,所以下面这篇文章主要给大家介绍了关于Python如何导入自己编写的py文件的相关资料,需要的朋友可以参考下

python二分法查找实例代码

二分算法是一种效率比较高的查找算法,其输入的是一个有序的元素列表,如果查找元素包含在列表中,二分查找返回其位置,否则返回NONE,下面这篇文章主要给大家介绍了关于python二分法查找的相关资料,需要的朋友可以参考下