Discuss / Python / po一个win10完全能用的代码 参考了https://blog.kasora.moe/2016/06/12/python-分布式计算/
Back

po一个win10完全能用的代码 参考了https://blog.kasora.moe/2016/06/12/python-分布式计算/

Topic source

影落天涯随风_635

#1 Created at ... [Delete] [Delete and Lock User]

master.py

import time,queue,randomfrom multiprocessing.managers import BaseManagerfrom multiprocessing import freeze_support#任务个数task_number = 10;#定义收发队列task_queue = queue.Queue(task_number);result_queue = queue.Queue(task_number);def gettask():    return task_queuedef getresult():     return result_queueclass QueueManager(BaseManager):    passdef test():    # 从BaseManager继承的QueueManager:    # 把两个Queue都注册到网络上, callable参数关联了Queue对象:    QueueManager.register('get_task_queue', callable=gettask)    QueueManager.register('get_result_queue', callable=getresult)    # 绑定端口5000, 设置验证码'abc':    manager = QueueManager(address=('127.0.0.1', 5002), 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.')if __name__ == '__main__':    #windows下多进程可能会炸，添加这句可以缓解    freeze_support()    test()

最后几行比较重要，注释说的很清楚，不加的话运行会报错，让你添加freez_support（）。

worker.py

import time, sys, queuefrom 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=('127.0.0.1', 5002), 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.')if __name__ == '__main__':    pass;

先运行master.py。再运行worker.py。再返回master查看结果，ok，perfect。

影落天涯随风_635

#2 Created at ... [Delete] [Delete and Lock User] 
import time,queue,randomfrom multiprocessing.managers import BaseManagerfrom multiprocessing import freeze_support#任务个数task_number = 10;#定义收发队列task_queue = queue.Queue(task_number);result_queue = queue.Queue(task_number);def gettask():    return task_queuedef getresult():     return result_queueclass QueueManager(BaseManager):    passdef test():    # 从BaseManager继承的QueueManager:    # 把两个Queue都注册到网络上, callable参数关联了Queue对象:    QueueManager.register('get_task_queue', callable=gettask)    QueueManager.register('get_result_queue', callable=getresult)    # 绑定端口5000, 设置验证码'abc':    manager = QueueManager(address=('127.0.0.1', 5002), 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.')if __name__ == '__main__':    #windows下多进程可能会炸，添加这句可以缓解    freeze_support()    test()

直接粘贴会有问题。。不换行

影落天涯随风_635

#3 Created at ... [Delete] [Delete and Lock User] 
import time,queue,randomfrom multiprocessing.managers import BaseManagerfrom multiprocessing import freeze_support#任务个数task_number = 10#定义收发队列task_queue = queue.Queue(task_number)result_queue = queue.Queue(task_number)def gettask():    return task_queuedef getresult():     return result_queueclass QueueManager(BaseManager):    passdef test():    # 从BaseManager继承的QueueManager:    # 把两个Queue都注册到网络上, callable参数关联了Queue对象:    QueueManager.register('get_task_queue', callable=gettask)    QueueManager.register('get_result_queue', callable=getresult)    # 绑定端口5000, 设置验证码'abc':    manager = QueueManager(address=('127.0.0.1', 5002), 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.')if __name__ == '__main__':    #windows下多进程可能会炸，添加这句可以缓解    freeze_support()    test()

try again....

影落天涯随风_635

#4 Created at ... [Delete] [Delete and Lock User]

不用代码格式了。。没有换行

master.py

import time,queue,random

from multiprocessing.managers import BaseManager

from multiprocessing import freeze_support

#任务个数

task_number = 10

#定义收发队列

task_queue = queue.Queue(task_number)

result_queue = queue.Queue(task_number)

def gettask():

    return task_queue

def getresult():

     return result_queue

class QueueManager(BaseManager):

    pass

def test():

    # 从BaseManager继承的QueueManager:

    # 把两个Queue都注册到网络上, callable参数关联了Queue对象:

    QueueManager.register('get_task_queue', callable=gettask)

    QueueManager.register('get_result_queue', callable=getresult)

    # 绑定端口5000, 设置验证码'abc':

    manager = QueueManager(address=('127.0.0.1', 5002), 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.')

if __name__ == '__main__':

    #windows下多进程可能会炸，添加这句可以缓解

    freeze_support()

    test()

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=('127.0.0.1', 5002), 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.')

if __name__ == '__main__':
pass;

影落天涯随风_635

#5 Created at ... [Delete] [Delete and Lock User]

master.py

import time,queue,random

from multiprocessing.managers import BaseManager

from multiprocessing import freeze_support

#任务个数

task_number = 10

#定义收发队列

task_queue = queue.Queue(task_number)

result_queue = queue.Queue(task_number)

def gettask():

    return task_queue

def getresult():

     return result_queue

class QueueManager(BaseManager):

    pass

def test():

    # 从BaseManager继承的QueueManager:

    # 把两个Queue都注册到网络上, callable参数关联了Queue对象:

    QueueManager.register('get_task_queue', callable=gettask)

    QueueManager.register('get_result_queue', callable=getresult)

    # 绑定端口5000, 设置验证码'abc':

    manager = QueueManager(address=('127.0.0.1', 5002), 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.')

if __name__ == '__main__':

    #windows下多进程可能会炸，添加这句可以缓解

    freeze_support()

    test()

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=('127.0.0.1', 5002), 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.')

if __name__ == '__main__':
pass;

  • 1

Reply