Day 32 process&threading_4

线程和进程 4

一、multiprocessing模块    

　　multiprocessing包是Python中的多进程管理包。

与threading.Thread类似，它可以利用multiprocessing.Process对象来创建一个进程。该进程可以运行在Python程序内部编写的函数。该Process对象与Thread对象的用法相同，也有start(), run(), join()的方法。

1, python的Process类的进程调用

##############  进程调用
#  调用进程模块和时间模块，并测试多进程功能

from multiprocessing import Process
import time

def pp(name):
    """
    函数：完成进程调用测试工作
    :return:
    """
    print("you are coming!",name,time.ctime())
    time.sleep(2)

if __name__ == '__main__':
    p_list = []

    # 开三个分进程，程序本身算主进程
    for i in range(3):
        p = Process(target=pp,args=("%s" %i,))  # 调用方式基本和thread相似
        p_list.append(p)
        p.start()

    for j in p_list:  # 分别阻断进程和主进程间关系
        j.join()

    print("end!",time.ctime())

2，继承类的调用方式

######## 继承Process类调用
#

from multiprocessing import Process
import time

class MyProcess(Process):
    """
    继承父类Process的所用属性和功能
    """
    def __init__(self):
        Process.__init__(self)  # 继承父类的所有__init__属性

    # 实例执行start时，自动触发run的执行，可查看系统源码追溯到
    def run(self):
        print("Welcome to beijing!",self,time.ctime())
        time.sleep(2)

if __name__ == '__main__':
    p_list = []
    for i in range(3):
        xx = MyProcess()
        xx.start()
        p_list.append(xx)

    for j in p_list:
        j.join()

    print("END!",time.ctime())

二、进程间的通讯

2、1  进程队列queue

from multiprocessing import Process,Queue

def xx(q,n):
    q.put(n*n+6)
    # 测试子线程队列的位置
    print("son process of:",id(q))

if __name__ == '__main__':
    q = Queue()
    # 测试主线程队列的位置
    print("main process of: ",id(q))

    for i in range(3):
        p = Process(target=xx,args=(q,i,))
        p.start()

    print(q.get())
    print(q.get())
    print(q.get())

'''
# 事实是为了证明：Queue虽然实现了线程间的交流，但是实际是在不同线程开辟了不一样的内存空间。然而linux优化，结果就如下了：mac是一样的
main process of:  4319925192
son process of: 4319925192
6
son process of: 4319925192
7
son process of: 4319925192
10
'''

2、2 进程管道 Pipe

############## Pipe 进程管道
"""
Pipe（）返回的两个连接对象代表管道的两端。
    每个连接对象都有send()和recv()方法（等等）。
    请注意，如果两个进程（或线程）尝试同时读取或写入管道的同一端，管道中的数据可能会损坏。
"""

from multiprocessing import Process,Pipe

def x_l(l_conn):
    """
    开辟一个进程负责lc的通话
    :param l_conn:
    :return:
    """
    l_conn.send("Welcome to beijing!")
    response = l_conn.recv()
    print("x:",response)
    l_conn.close()

if __name__ == '__main__':
    x_conn,l_conn = Pipe()
    p = Process(target=x_l,args=(l_conn,))  # 开子进程，负责lc
    p.start()
    res = x_conn.recv()
    print("l:",res)
    x_conn.send("Im coming!")
    p.join()

"""
l: Welcome to beijing!
x: Im coming!
"""

 2，3 Manager 数据共享

############## Manager 数据共享:一个数据去更改另一个进程里的数据

from multiprocessing import Manager,Process

def xl(Mlist,i):
    Mlist.append(i)

if __name__ == '__main__':
    manager = Manager()  # 实例一个Manager
    Mlist = manager.list([1,"a"])  # 数据共享类型为列表，也可以用字典等
    l = []

    # 开子进程，并往主线程共享列表添加变量
    for i in range(5):
        p = Process(target=xl,args=(Mlist,i,))
        p.start()
        l.append(p)

    for j in l:
        j.join()

    print(Mlist

2，4 Pool 进程池

############## Pool
"""
进程池内部维护一个进程序列，当使用时，则去进程池中获取一个进程，
如果进程池序列中没有可供使用的进进程，那么程序就会等待，
直到进程池中有可用进程为止。
"""
from multiprocessing import Pool
import time

def xl(n):
    print(n)
    time.sleep(2)
    print("END!")

if __name__ == '__main__':
    pool_obj = Pool()  # 进程池，默认进程数是cpu核数，其中os.cpu_count()查看
    for i in range(20):
        pool_obj.apply_async(func=xl,args=(i,))
    pool_obj.close()  # 执行后，不会有新等进程进入进程池
    pool_obj.join()  # join（）在close（）之后，套路，牢记！！！

    print("ALL IS OVER!")

p.apply(func [, args [, kwargs]]):在一个池工作进程中执行func(*args,**kwargs),然后返回结果。需要强调的是：此操作并不会在所有池工作进程中并执行func函数。如果要通过不同参数并发地执行func函数，必须从不同线程调用p.apply()函数或者使用p.apply_async()
p.apply_async(func [, args [, kwargs]]):在一个池工作进程中执行func(*args,**kwargs),然后返回结果。此方法的结果是AsyncResult类的实例，callback是可调用对象，接收输入参数。当func的结果变为可用时，将理解传递给callback。callback禁止执行任何阻塞操作，否则将接收其他异步操作中的结果。
   
p.close():关闭进程池，防止进一步操作。如果所有操作持续挂起，它们将在工作进程终止前完成5 P.jion():等待所有工作进程退出。此方法只能在close（）或teminate()之后调用

三， 协程

"""
协程是一种用户态的轻量级线程，即协程是由用户程序自己控制调度的。
对比操作系统控制线程的切换，用户在单线程内控制协程的切换，优点如下：
1.  协程的切换开销更小，属于程序级别的切换，操作系统完全感知不到，因而更加轻量级
2. 单线程内就可以实现并发的效果，最大限度地利用cpu
"""

import time

"""
传统的生产者-消费者模型是一个线程写消息，一个线程取消息，通过锁机制控制队列和等待，但一不小心就可能死锁。
如果改用协程，生产者生产消息后，直接通过yield跳转到消费者开始执行，待消费者执行完毕后，切换回生产者继续生产，效率极高。
"""
# 注意到consumer函数是一个generator（生成器）:
# 任何包含yield关键字的函数都会自动成为生成器(generator)对象

def consumer():
    r = ''
    while True:
        # 3、consumer通过yield拿到消息，处理，又通过yield把结果传回；
        #    yield指令具有return关键字的作用。然后函数的堆栈会自动冻结(freeze)在这一行。
        #    当函数调用者的下一次利用next()或generator.send()或for-in来再次调用该函数时，
        #    就会从yield代码的下一行开始，继续执行，再返回下一次迭代结果。通过这种方式，迭代器可以实现无限序列和惰性求值。
        n = yield r
        if not n:
            return
        print('[CONSUMER] ←← Consuming %s...' % n)
        time.sleep(1)
        r = '200 OK'
def produce(c):
    # 1、首先调用c.next()启动生成器
    next(c)
    n = 0
    while n < 5:
        n = n + 1
        print('[PRODUCER] →→ Producing %s...' % n)
        # 2、然后，一旦生产了东西，通过c.send(n)切换到consumer执行；
        cr = c.send(n)
        # 4、produce拿到consumer处理的结果，继续生产下一条消息；
        print('[PRODUCER] Consumer return: %s' % cr)
    # 5、produce决定不生产了，通过c.close()关闭consumer，整个过程结束。
    c.close()
if __name__=='__main__':
    # 6、整个流程无锁，由一个线程执行，produce和consumer协作完成任务，所以称为“协程”，而非线程的抢占式多任务。
    c = consumer()
    produce(c)
    
    
'''
result:

[PRODUCER] →→ Producing 1...
[CONSUMER] ←← Consuming 1...
[PRODUCER] Consumer return: 200 OK
[PRODUCER] →→ Producing 2...
[CONSUMER] ←← Consuming 2...
[PRODUCER] Consumer return: 200 OK
[PRODUCER] →→ Producing 3...
[CONSUMER] ←← Consuming 3...
[PRODUCER] Consumer return: 200 OK
[PRODUCER] →→ Producing 4...
[CONSUMER] ←← Consuming 4...
[PRODUCER] Consumer return: 200 OK
[PRODUCER] →→ Producing 5...
[CONSUMER] ←← Consuming 5...
[PRODUCER] Consumer return: 200 OK
'''

 3,1 Gevent

#############   协程 greenlet

from gevent import monkey
monkey.patch_all()
import gevent
from urllib import request
import time

def xl(url):
    print("GET:%s" %url)
    res = request.urlopen(url)
    data = res.read()
    print("%d bytes recevied from %s" %(len(data),url))

start = time.time()

gevent.joinall([
    # gevent.spawn(xl,"http://www.xiaohuar.com"),
    gevent.spawn(xl,"http://www.mmjpg.com"),
    gevent.spawn(xl,"http://www.fengniao.com")
])

print(time.time()-start)

 

 

Greenlet

Gevent 自动切换的牛逼之处，可以的就开干