python成长之路 多进程、协程

一、python3的多进程

1、multiprocessing模块

python3是通过multiprocessing模块来开启子进程，并执行相应的定制任务

multiprocessing模块可以支持子进程、通信和数据共享、执行不同形式的同步，提供了process、Queue、Pipe、lock等组件。

在这里我们强调的是多线程有共享状态，进程没有任何共享状态，

2、Process类的介绍

• 创建进程的类

我们通过Process类来创建子进程。下面是关于Process类中的参数的解释

Process(group=None, target=None, name=None, args=(), kwargs={})

在这些参数中group参数未使用，始终未None

target相当于我们要开启的子进程的任务的位置也就是调用的对象

args使我们给这个子进程传递参数，这个必须是元组的方式，如果只有一个参数就写成(num,)这种形式记得后面有一个“,” args = (num,)

kwargs表示调用对象的字典，kwargs={"name":"Tom","age":18}

name为子进程的名称

记住上面只是创建一个子进程这个子进程属于未开启状态

• 方法介绍

如上面我盟创建了一个子进程如

p = multiprocessing.Process()  #这个只是创建了一个子进程但是他并没有运行。

使用p.start(): 这个表示启动这个子进程，并调用了该子进程的p.run()

p.run(): 进程启动时的方法，正是它去调用target指定的函数，在multiprocessing里面已经实现，要是我们自定义类的话一应要实现该方法。

p.is_alive():主要是看p是否在运行，如果还在运行那么返回True。

p.join([timeout]): 主进程等待子进程p执行完毕才会执行，所以在p执行完毕之前主进程处于等待状态。timeout是可选的超时时间，在这里我们要记住p.join只能等待start开启的进程，而对run开启的进程没有效果。

• 一些属性介绍

p.daemon:默认值为Flash，如果设为True,代表p为后台云逊的守护进程，当p的父进程终止时，p也随之终止，并且设置为True后，p不能创建自己的子进程，必须在p.start()之前设置。

p.name：进程的名称。

p.pid: 进程的pid

3、 使用Process类创建进程

#!/usr/bin/env python
# -*-coding:utf-8-*-
import multiprocessing
import time


def run(id):
    time.sleep(2)
    print("hello process{}".format(id))

if __name__ == "__main__":
    l = []
    i = 1
    for i in range(5):
        p = multiprocessing.Process(target=run, args=(i,))#args是一个元组所以必须要有，
        p.start()
        i += 1
        l.append(p)
    for i in l:     
        i.join() 
    print("主进程")
#如果没有i.join()的话就会出现子进程没有运行完，下面的代码就会运行结果。

上面的运行结果是：
hello process1
hello process0
hello process4
hello process2
hello process3
主进程


如果没有i.join()的话，运行结果为
主进程
hello process0
hello process3
hello process4
hello process1
hello process2

从上可以看出当有i.join()时我们的程序就不会往下走。它会等待join()过的子进程执行完毕才会继续往下走

　　

下面的演示是关于在子进程中开启多线程

#!/usr/bin/env python
# -*-coding:utf-8-*-

import multiprocessing,time
import threading


def run():
    time.sleep(2)
    print("hello process")
    t = threading.Thread(target=trun,)
    t.start()


def trun():
    time.sleep(1)
    print("hello threading", threading.get_ident())


def main():
    l = []
    for i in range(10):
        p = multiprocessing.Process(target=run)
        p.start()
        print(p.name)
        l.append(p)
    for i in l:
        i.join()


if __name__ == "__main__":
    main()

#在Windows中的process()执行必须放到if __name__ == '__main__':下如上面的代码，要不然就会出错。

　　

关于p.daemon=True的用法

#!/usr/bin/env python
# -*-coding:utf-8-*-
import multiprocessing
import time


def run(id):
    time.sleep(2)
    print("hello process{}".format(id))

if __name__ == "__main__":
    l = []
    i = 1
    for i in range(5):
        p = multiprocessing.Process(target=run, args=(i,))
        p.daemon = True   #记住这个一定要在start()方法之前设置
#同时记住当我们设置守护进程时p进程就不能创建子进程，并且父进程结束，p也结束不管p有没有运行完
        p.start()
        i += 1
        l.append(p)
    for i in l:
        i.join()  #去掉这个后就会发现开启守护进程的区别

    print("主进程")

#上面的加上守护进程我们不会发现什么不同和我们前面没加上一样的结果。
#但是当我们去掉join()方法的时候就会发现子进程还没有运行完毕整个进程就结束了

　　

4、进程同步（锁）

上面我们也说了进程之间的数据是不共享的，但是共享同一套文件系统，所以访问同一个文件或者同一个打印终端是没问题的。当我们同一个终端打印的时候会发现多行打印到一行的现象，出现这种现象就是多个进程共享并抢占同一个终端打印导致打印乱了。

在这里我们加锁的就是为了保证同一时间同一个数据只能被一个进程修改，这样做速度慢了但是却也保证了数据的正确性

#!/usr/bin/env pytho# 
-*-coding:utf-8-*

from multiprocessing import Process,Lock

def run(l, i):
    l.acquire()
    print("process:", i)  # 在屏幕上打印的时候打印完整
    l.release()

if __name__ == "__main__":
    lock = Lock()
    p_list = []
    for i in range(10):
        p = Process(target=run, args=(lock, i))
        p.start()
        p_list.append(p)
    for i in p_list:
        i.join()

运行结果为:
process: 1
process: 0
process: 2
process: 8
process: 7
process: 5
process: 3
process: 6
process: 4
process: 9

这个程序我运行了很多次都没有出现过打印错乱的情况，
比如第一行没打印完，第二行就插在中间的这种情况。
不过我我在Linux上弄了好多次包括python3和python2
都没有出现过相应的情况。有可能我试验少了，但是看别
人的博客上说python2运行的时候有可能出现错乱的现象。
不管怎么样记住加上进程锁就是为了防止在屏幕上打印错乱。

　　

5、进程间的通讯

队列 Queue

上面我们说了不同的进程是不能够共享内存的，但是为了实现两个进程之间的数据交换我们可以使用一下方法。

Queue

这里使用的queue和线程threading的queue的用法差不多。

队列属于先进先出的原则的

创建队列的类：

Queue([maxsize]): 创建共享的进程队列，Queue是多进程安全的队列，可以使用Queue实现多进程之间的数据传递。

关于里面的maxsize表示的是队列允许的最大项数，省略表示无大小限制。

关于Queue的方法介绍：

q.put 是用来向队列中插入数据的

q.get 是向队列中去除一个元素，并删除掉

q.empty():调用此方法时，q为空则返回True 

q.full():调用此方法时q已满则返回True

q.qsize():返回队列中目前元素的数目

#!/usr/bin/env python
# -*-coding:utf-8-*-
from multiprocessing import Process, Queue


def run(arg):
    arg.put([41, None, 'miss'])

if __name__ == "__main__":
    q = Queue()
    p = Process(target=run, args=(q,))
    p.start()
    print("进程:", q.get())
    p.join()


运行的结果为
进程: [41, None, 'miss']

　　

管道 Pipe

 进程间的通信方式除了队列还有管道。

创建管道的类：

Pipe([duplex]):在进程间创建一条管道，并返回元组（conn1, conn2）,其中conn1,conn2表示管道两端的连接对象，必须在产生process对象之前，产生管道。 

duplex：默认管道是双全工，如果duplex设置为Flash那么conn1只能用于接受。conn2只能用于发送

关于Pipe其中方法的介绍：

conn1.recv():接收conn2.send(obj)发送的对象。如果没有消息可接受，recv会一直阻塞，如果连接的另一端关闭了那么recv方法会抛出EOFError。

conn2.send(obj):通过连接发送对象，obj是与序列化兼容的对象

#!/usr/bin/env python
# -*-coding:utf-8-*-
from multiprocessing import Process, Pipe


def run(arg):
    arg.send([1, None, "chengshun"])
    arg.send([2, None, "chengjie"])

if __name__ == "__main__":
    parent_conn, child_conn = Pipe()  在创建进程之前使用Pipe()方法
    p = Process(target=run, args=(child_conn,))
    p.start()
    print(parent_conn.recv())
    print(parent_conn.recv())
    p.join()

　　

manager

主要是实现在进程间实现list，dict，Lock等等这种类型的数据的共享

#!/usr/bin/env python
# -*-coding:utf-8-*-
from multiprocessing import Process, Manager
import os


def run(d, l):
    d[os.getpid()] = os.getpid()
    l.append(os.getpid())
    #print(l)

if __name__ == "__main__":
    with Manager() as manager:  # 把Manager命名为manager
        d = manager.dict()  # 创建一个共享的字典d
        l = manager.list(range(10))  # 创建一个共享的列表l
        process = []
        for i in range(10):
            p = Process(target=run, args=(d, l))  # 创建进程
            p.start()
            process.append(p)
        for i in process:
            i.join()

        print(l)
        print(d)

　　

6、进程池Pool

我们开多进程就是为了并发执行任务，当我们有多少核就开多少个进程，但是在日程中我们常常需要的并发执行任务远大于核数，这个是后我们就可以通过维护一个进程池来控制进程的数目。

进程池内部维护一个进程序列，当时用的，则取进程池中获取一个进程，如果进程池中没有可用序列的进程，那么程序就会等待知道进程池中有可用的进程序列为止。

Pool对象调用join方法会等待所有的子进程执行完毕，

在调用join方法之前，必须调用close

调用close之后就不能够继续添加新的process了

Pool的中2中方法

apply_async:该方法用来同步执行进程，也就是说允许同时多个进程进入到进程池中。

apply:该方法只能允许一个进程进入池子，

#!/usr/bin/env python
# -*-coding:utf-8-*-
from multiprocessing import Pool,process
import os, time


def Foo(i):
    time.sleep(1)
    print("子进程号:", os.getpid())
    return i + 100


def back(arg):  # 这个回调函数是主进程调用的
    print("---exit---", arg)  # 这里的arg就是func中的return返回值，如果没有，那么arg为None


if __name__ == "__main__":
    pool = Pool(processes=5)  # 进程池中同时可以允许多少个进程
    for i in range(40):
        pool.apply_async(func=Foo, args=(i,), callback=back)  # callback为回调函数
        # pool.apply(func=Foo, args=(i,))  
        # 也就是说在没次执行完func这个函数的时候就会调用callback函数。

    pool.close()
    pool.join()  # join()之前必须要close()，同时close()后就不能够在添加新的process
   # 如果直接注释掉pool.join()那么程序就不会等待子进程结束就自动结束了

　　

上面代码中的pool.apply_async()这个运行的结果我们会看见会出现运行的时候去掉---exit---会出现5个5个一组打印在屏幕上。如果是pool.apply()这个运行的时候我们会发现会一行一行的打印。也就是串行的结果。

二、协程

协程，又称为微线程，纤程。英文名字Coroutine。我们可以说协程是一种用户态的轻量级线程。

线程是系统级别的它由操作系统调度，而协程则是应用程序级别的由程序根据需求自己调度。一个线程中会有很多函数，我们把这些函数称之为子程序，在子程序执行过程中可以去执行别的子程序，而别的子程序也可以中断回来继续执行之前的子程序，这个过程我么称之为协程。也就是说在同一线程内一段代码在执行过程中会中断然后跳转执行别的代码，接着在之前中断的地方继续执行，类似于yield操作。

协程拥有自己的寄存器上下文和栈。协程调度切换时，将寄存器上下文和栈保存到其他地方，在切回来的时候，恢复先前保存的寄存器上下文和栈。因此：协程能保留上一次调用时的状态（即所有局部状态的一个特定组合），每次过程重入时，就相当于进入上一次调用的状态，换种说法：进入上一次离开时所处逻辑流的位置

协程的优点：

        （1）无需线程上下文切换的开销，协程避免了无意义的调度，由此可以提高性能（但也因此，程序员必须自己承担调度的责任，同时，协程也失去了标准线程使用多CPU的能力）

　　（2）无需原子操作锁定及同步的开销

　　（3）方便切换控制流，简化编程模型

　　（4）高并发+高扩展性+低成本：一个CPU支持上万的协程都不是问题。所以很适合用于高并发处理。

协程的缺点：

　　（1）无法利用多核资源：协程的本质是个单线程,它不能同时将 单个CPU 的多个核用上,协程需要和进程配合才能运行在多CPU上.当然我们日常所编写的绝大部分应用都没有这个必要，除非是cpu密集型应用。

　　（2）进行阻塞（Blocking）操作（如IO时）会阻塞掉整个程序

1、关于yield实现协程效果

import time
import queue


def consumer(name):
    print("--->开始吃包子..." )
    while True:
        print("%s 需要包子" % name)
        new_baozi = yield
        print("[%s] 吃了包子%s" % (name, new_baozi))


def producer():
    r = con.__next__()
    r = con2.__next__()
    n = 0
    while n < 5:
        n += 1
        con.send(n)
        con2.send(n)
        print("33[32;1m[producer]33[0m 是包子 %s" % n)


if __name__ == '__main__':
    con = consumer("c1")
    con2 = consumer("c2")
    p = producer()

　　

2、greenlet模块实现模块之间的切换

#!/usr/bin/env python
# -*-coding:utf-8-*-
from greenlet import greenlet


def fun1():
    print("1")   # 程序先执行这一行
    gre2.switch()  # 这表示我们切换到gre2，也就是fun2.同时保留现在这个状态，下次切回继续执行
    print(4)  # 有下面的gre1.switch()切换到这执行该行
    gre2.switch()  # 在切换到gre2


def fun2():
    print("2")  # 由gre2.switch()切换到这执行
    gre1.switch()  # 这个表示切换到上面gre1，如果上面有保留状态则从保留状态开始
    print(3)


if __name__ == "__main__":
    gre1 = greenlet(fun1)  # 启动一个协程
    gre2 = greenlet(fun2)  # 启动一个协程
    gre1.switch()  # 一开始运行gre1也就是fun1

　　上面的模块我们可以知道这个需要我们手动切换，不能够实现IO自动切换。

3、gevent

gevent是一个第三方库，可以轻松通过gevent实现协程编程，在gevent用到的主要模块时greenlet，它是以C扩展模块形式接入到Python的轻量级协程。 greenlet全部运行在主程序操作系统的内部，但它们协作式的调度。下面我们看看

#!/usr/bin/env python
# -*-coding:utf-8-*-
import gevent


def f1():
    print("1")  
    gevent.sleep(2)  # 到这里程序会自动的切换到f2函数
    print("6")


def f2():
    print("2")
    gevent.sleep(1) # 自动切换到f3哪怕这个sleep了0秒，它也会切换到下一个，相当于触发了自动切换。
    print("5")


def f3():
    print("3")
    gevent.sleep(0) # 这个会切换到f1看看有没有运行完，如果没有运行完就按着顺序继续运行下去。
    print("4")

if __name__ == "__main__":
    gevent.joinall([
        gevent.spawn(f1),  # 相当于启动一个协程，第一位就是一开始运行
        gevent.spawn(f2),
        gevent.spawn(f3)
    ])

　上面的程序运行的时候我们可以看到运行的结果是

1

2

3

4

5

6

它基本的实现了IO的自动切换

下面一个例子时候关于协程，为了体现差异我们同时也执行了同步时间

#!/usr/bin/env python
# -*-coding:utf-8-*-
import gevent, time
from urllib import request  # 包下面还有一个包，所以必须这样不能够使用urllib.request
from gevent import monkey


def run(url, name):
    res = request.urlopen(url)
    data = res.read()
    f = open('{}.txt'.format(name), 'wb')
    f.write(data)
    f.close()
    print("{} bytes received from {}".format(len(data), url))


if __name__ == "__main__":
    # 同步时间
    get = [
        ("https://www.python.org", 'python'),
        ("https://www.yahoo.com", 'yahoo'),
        ("https://www.github.com", 'github')  ]
    start_time = time.time()
    for i, j in get:
        run(i, j)
    done_time = time.time()
    print("同步时间", done_time - start_time)

    # 异步时间
    monkey.patch_all()  # urllib默认情况下是串行的使用这个是把当前程序的所有的IO操作的打上标记这个样就可以异步了
    asy_start_time = time.time()
    gevent.joinall([
        gevent.spawn(run, "https://www.python.org", 'python'),
        gevent.spawn(run, "https://www.yahoo.com", 'yahoo'),
        gevent.spawn(run, "https://www.github.com", 'github')

    ])  # 从第一位开始运行
    asy_done_time = time.time()
    print("异步时间", asy_done_time - asy_start_time)

运行的结果是
48872 bytes received from https://www.python.org
532554 bytes received from https://www.yahoo.com
52840 bytes received from https://www.github.com
同步时间 6.285885572433472
48872 bytes received from https://www.python.org
525992 bytes received from https://www.yahoo.com
52840 bytes received from https://www.github.com
异步时间 3.48797869682312

这个有时候时间是不对的，因为和网速有关，这几个网址有时候会出现打开速度慢。

　　

4、协程实现socket多并发

gevent_socket_server

#!/usr/bin/env python
# -*-coding:utf-8-*-
import socket, gevent
from gevent import monkey
monkey.patch_all()  # 必须要有这个，要不然就实现多并发


def Server():
    server = socket.socket()
    server.bind(("localhost", 9999))
    server.listen(500)
    while True:
        conn, add = server.accept()
        gevent.spawn(handle_request, conn)


def handle_request(conns):
    while True:
        received_data = conns.recv(1024).decode("utf-8")
        print("received:", received_data)
        if not received_data:
            conns.shutdown(socket.SHUT_WR)
        conns.send(received_data.encode("utf-8"))
    conns.close()

if __name__ == "__main__":
　　Server()

　

socket_client

#!/usr/bin/env python
# -*-coding:utf-8-*-
import socket

client = socket.socket()
client.connect(("localhost", 9999))
while True:
    acquire = input(">>>").strip().encode("utf-8")
    print(acquire)
    if acquire.decode("utf-8") == "exit":
        break
    if len(acquire) == 0:
        continue
    client.send(acquire)
    print(2)
    data = client.recv(1024).decode("utf-8")
    print(3)
    print(data)
client.close()

　这个就是记录自己学习中的知识，同时也知道写的也不是很全。这个权当自己对知识的巩固。