Python多进程

from multiprocessing import Process
import time,random
num = 258


class MyPro(Process):
    def run(self):
        time.sleep(10)
        print(‘%s子进程退出‘ % self.name)


if __name__ == ‘__main__‘:
    m1 = MyPro()
    m2 = MyPro()
    m1.start()
    m2.start()
    m1.join()
    m2.join()
    print(‘主进程的主线程退出‘)

# 注意：
　　1.多进程的join()和多线程的join()卡住的不是进程，而是主线程,谁调用了join,就等于告诉主线程:我执行完了，你主线程才能往下执行。
　　　　另外，子进程也是有自己的主线程的
　　2.所有子线程或者子进程集中调用了join后，主线程的等待的总时间以 耗费时间最长的那个进程或者线程的运行时间 为准
　　3.如果你创建了一个子进程或者线程运行后立马跟着一个join(),会导致串行执行，比如以上代码改成：
　　　　m1.start()
　　　　m1.join()
　　　　m2.start()
　　　　m2.join()
　

from multiprocessing import Process
import time,random
num = 258


class MyPro(Process):
    def run(self):
        time.sleep(1)
        print(‘%s子进程退出‘ % self.name)


if __name__ == ‘__main__‘:

    m1 = MyPro()
    m1.daemon = True # 设置守护进程，要在start()之前设置。主进程执行退出，无论子进程执行到哪里，都跟着主进程退出
    m1.start()
    print(‘主进程退出‘)

# 每个进程内部的数据是独立的，但是，当共同访问一个文件、数据库、打印机、其他设备等是会带来竞争的。
# 解决的办法是加锁，但这种做法会牺牲效率，并发执行变成了串行执行

如：3个进程，要排队上公共汽车，过程应该是，一个先上完车，下一个跟进

from multiprocessing import Process,Lock
import time
class MyPro(Process):
    def run(self):
        print(‘%s在排队上车‘ % self.name)
        time.sleep(2)
        print(‘%s上车完毕‘ % self.name)

if __name__ == ‘__main__‘:
    pro_list = []
    for i in range(3):
        m = MyPro()
        m.start()
# 结果：

MyPro-2在排队上车
MyPro-1在排队上车
MyPro-3在排队上车
MyPro-2上车完毕
MyPro-1上车完毕
MyPro-3上车完毕

#得到的结果是 三个同时挤上车

from multiprocessing import Process, Lock
import time
lock = Lock()      # 生成一个锁对象
class MyPro(Process):
    def __init__(self, lock):
        super().__init__()
        self.lock = lock 

    def run(self):
        self.lock.acquire() # 谁拿到锁，就有执行权
        print(‘%s正在排队上车‘ % self.name)
        time.sleep(2)
        print(‘%s已经上车‘ % self.name)
        self.lock.release() # 释放锁，让其他进程去抢

if __name__ == ‘__main__‘:
    for i in range(3):
        m = MyPro(lock)
        m.start()

1.q.put(blocked,timeout): 将数据推入队列里。可选参数:blocked=True，timeout是正整数，该方法会阻塞到timeout的指定时间，阻塞到指定时间后，队列没有空间，抛出
　　　　　　　　　　　　　　　queue.Full() 异常；blocked=False,队列满了就会抛queue.Full()异常
2.q.get(blocked, timeout): 从队列里取出一个元素并删除。可选参数:blocked=True，timeout是正整数，这等待时间内没有取到元素，抛出queue.Empty()异常。
　　　　　　　　　　　　　　　　blocked=False,队列为空，抛出queue.Empty()异常
3.q.get_nowait():同q.get(False)
4.q.put_nowait():同q.put(False)

from multiprocessing import Process, Queue
import time


class Chef(Process):
    def __init__(self, queue):
        super().__init__()
        self.queue = queue

    def run(self):
        n = 5
        while n > 0:
            time.sleep(1)
            print(‘厨师生产了编号[-%s-]的包子‘ % n)
            self.queue.put(n)
            n -= 1


class Foodie(Process):
    def __init__(self, queue):
        super().__init__()
        self.queue = queue

    def run(self):
        while True:
            data = self.queue.get()
            if data == None:
                break
            print(‘食客[-%s-]吃了编号[-%s-]的包子‘ % (self.name, data))
            time.sleep(1)


if __name__ == ‘__main__‘:
    q = Queue()
    ch = Chef(q)
    ch.start()

    for i in range(3):
        fd = Foodie(q)
        fd.start()
    ch.join()  # 主线程等待生产者生产完成后才往下执行
    for i in range(3):  # 有多少个消费者就发送多少个None，接受到None的子进程就退出
        q.put(None)

JoinableQueue继承了Queue,拥有Queue之外的方法：
1.q.task_done():向q.join()发送信号，表示数据已经被取走
2.q.join() ：调用此方法，将会阻塞，直到 q.task_done()的调用

尽量使用消息队列进程并发编程，另外，应要避免共享数据的使用

#在远程控制多主机或者同时操作一些系统管理的时候，并行可以节省时间。另外，如果任务过多，频繁的创建进程太麻烦和繁琐，任务少可以选择不用进程池。

# 参数
from multiprocessing import Pool
Pool(self, processes=None, initializer=None, initargs=(),maxtasksperchild=None, context=None)
processes: 指定创建进程数量
initializer: 每个进程启动时要执行的可调用对象
initargs: 进程的参数，元祖

pool.apply(func, args=(), kwds={}) 同步执行，并返回结果
pool.apply_async(func, args=(), kwds={}, callback=None,error_callback=None) 
‘‘‘ 
异步执行，CPU有多少个核，就能同时跑多少个任务，参数callback代表当前一次的任务执行完毕后，通知主进程调用一个函数(回调)，另外,callback的函数不能有阻塞操作，因为会影响到其他异步任务的结果。
当进程池里的进程数量大于CPU核数和任务数量比进程池数大时，剩余的进程会切换着运行。如，进程池=5，CPU核数=4，任务数量=10，那么同时能跑4个任务，剩下的一个切换着跑，只是太快了感觉不出来
‘‘‘
pool.close() 关闭进程池
pool.join() 等待所有工作进程退出，另外，join必须要在close的前面

from multiprocessing import Pool
import os,time


def start(num):
    print(‘创建了PID为[-%s-]的进程‘ % os.getpid())
    time.sleep(1)
    print(‘进程[-%s-]执行完毕‘ % os.getpid())
    return num


def end(arg):
    print(arg)  # 打印每个任务的结果
    print(‘结束了‘, os.getpid())  # 查看是哪个进程打印(其实是Python解释器的进程)


if __name__ == ‘__main__‘:
    pool = Pool(2)  # 指定最大进程数
    result = []
    for i in range(10):  # 创建10个任务
        ‘‘‘
        当一个任务执行完毕后，通知主进程调用end, 参数为当前任务的返回结果
        ‘‘‘
        res = pool.apply_async(func=start, args=(i,), callback=end) 
        result.append(res)


    ‘‘‘
    apply_async()不同于apply(),异步执行，主进程要使用join()等待，等所有任务执行完毕使用get()来收集,
    如果不这么做，主进程结束，进程池没来的及执行，就跟着主进程一起结束了
    ‘‘‘
    pool.close()
    pool.join()
    for res in result:
        print(res.get())  # 获取结果，apply()可直接获取

from socket import *
from multiprocessing import Pool
import os

s = socket(AF_INET, SOCK_STREAM)
s.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)
s.bind((‘127.0.0.1‘, 888))
s.listen(5)


def Server_Pro(conn, addr):
    while True:
        data = conn.recv(1024)
        print(‘收到客户端发来的消息%s‘ % data)
        talk = ‘Server收到你发送过来的消息‘.encode(‘utf-8‘)
        conn.send(talk)


if __name__ == ‘__main__‘:
    pool = Pool(3)  # 能同时处理三个任务
    while True:
        conn, addr = s.accept()
        pool.apply_async(Server_Pro,args=(conn, addr))

from socket import *
c = socket(AF_INET, SOCK_STREAM)
c.connect((‘127.0.0.1‘, 888))
while True:
    data = input(‘输入发送的消息:‘).encode(‘utf-8‘)
    c.send(data)
    msg = c.recv(1024)
    print(msg.decode(‘utf8‘))