#Python分布式进程快速入门与实战（附完整代码）

import random
import time
from multiprocessing import Queue
from multiprocessing.managers import BaseManager

# 自定义管理器类（必须继承BaseManager）
class QueueManager(BaseManager):
    pass

def main():
    # 1. 初始化本地队列
    task_queue = Queue()   # 存待计算的随机数
    result_queue = Queue() # 存计算结果

    # 2. 把本地队列注册为网络可调用的接口
    # 注意：用lambda匿名函数返回队列实例，避免提前实例化问题
    QueueManager.register('get_task_queue', callable=lambda: task_queue)
    QueueManager.register('get_result_queue', callable=lambda: result_queue)

    # 3. 启动网络管理器
    # address: ('0.0.0.0', 端口) 表示监听本机所有网卡的该端口
    # authkey: 字节串密钥，Master和Worker必须完全一致，防止非法连接
    manager = QueueManager(address=('0.0.0.0', 56789), authkey=b'python_distributed_test_202x')
    manager.start()
    print(f"✅ Master启动成功！监听端口 56789")

    try:
        # 4. 获取代理队列对象（和本地Queue用法完全一致）
        proxy_task = manager.get_task_queue()
        proxy_result = manager.get_result_queue()

        # 5. 分发模拟任务：放10个0-10000的随机数
        print("\n📦 开始分发任务...")
        for _ in range(10):
            n = random.randint(0, 10000)
            print(f"→ 放入任务：计算 {n} 的平方")
            proxy_task.put(n)

        # 6. 阻塞等待Worker返回结果（设置30秒超时避免死等）
        print("\n⏳ 等待Worker返回结果...")
        for _ in range(10):
            res = proxy_result.get(timeout=30)
            print(f"← 收到结果：{res}")

    except KeyboardInterrupt:
        print("\n⚠️  Master被手动中断")
    finally:
        # 7. 清理资源，关闭网络管理器
        manager.shutdown()
        print("\n👋 Master已安全退出")

if __name__ == "__main__":
    main()

import time
import sys
from multiprocessing.managers import BaseManager

# 自定义管理器类（和Master保持一致，否则注册会失败）
class QueueManager(BaseManager):
    pass

def main():
    # 1. 注册要调用的Master接口（只需要名称，不需要具体实现）
    QueueManager.register('get_task_queue')
    QueueManager.register('get_result_queue')

    # 2. 从命令行或配置获取Master的IP（默认本机，方便单测）
    server_ip = sys.argv[1] if len(sys.argv) > 1 else '127.0.0.1'
    print(f"🔗 正在连接Master: {server_ip}:56789...")

    try:
        # 3. 连接到Master的网络管理器
        manager = QueueManager(address=(server_ip, 56789), authkey=b'python_distributed_test_202x')
        manager.connect()
        print("✅ 连接Master成功！")

        # 4. 获取代理队列对象
        proxy_task = manager.get_task_queue()
        proxy_result = manager.get_result_queue()

        # 5. 循环取任务（直到超时10秒没有新任务，自动退出）
        print("\n🏃 Worker开始工作...按Ctrl+C可手动停止")
        while True:
            try:
                n = proxy_task.get(timeout=10)
                print(f"→ 收到任务：计算 {n} 的平方")
                # 模拟1秒的真实耗时任务
                time.sleep(1)
                res = f"{n} * {n} = {n*n}"
                print(f"← 完成任务：{res}")
                proxy_result.put(res)
            except Exception as e:
                print(f"\n⚠️  Worker异常/无新任务：{str(e)}")
                break

    except ConnectionRefusedError:
        print("❌ 连接失败：请检查Master是否启动、IP/端口/密钥是否正确")
    finally:
        print("\n👋 Worker已安全退出")

if __name__ == "__main__":
    main()