day11 消息队列、多线程、多进程、协程

阅读量：6619 次

发布时间：2019-06-25

本文共 21322 字，大约阅读时间需要 71 分钟。

Python 消息队列、多线程、多进程、协程（二）

本章内容简介

1. queue 消息队列详解：常用参数、队列的四种类型及实例；

2. Python 多线程：线程池、线程锁

3. Python 多进程: 进程锁、进程池、数据共享

4. 协程

一. queue 消息队列

1）常用参数：

a = queue.Queue() #括号里是可以放几个数据，2就是2个；多的就放不进去了

a．put(11) # put放数据时，可以设置是否阻塞，超时时间

a．put(22)

a.get() # 取数据，按模式取值，默认（阻塞）；可以设置参数（block=False），超时时间（timeout=2）2秒；

a.qsize() 　　　　　# 计数，列队里有几个值；

a.task_done()　　　 # 标识列队里的此任务已经完成了，和join配合使用；

a.join() 　　　　　　 # 如果列队没执行完，程序不终止，执行完要告诉队列，执行完了，才能终止操作。

2）队列的四种模式：

# queue.Queue，先进先出队列

# queue.LifoQueue，后进先出队列

# queue.PriorityQueue，优先级队列

# queue.deque，双向对队

实例：

import queue# q = queue.Queue(2) 先进先出队列# q = queue.LifoQueue() 后进先出队列# q = queue.PriorityQueue() 优先级队列# q = queue.deque()   双向队列q = queue.LifoQueue()q.put(123)q.put(456)# 打印；456print(q.get())# 优先级最小的拿出来# 如果优先级一样，则是谁先放，就先取出谁q = queue.PriorityQueue()q.put((1,'alex1'))q.put((1,'alex2'))q.put((1,'alex3'))q.put((3,'alex3'))# (1, 'alex1')print(q.get())q = queue.deque()q.append(123)q.append(333)q.appendleft(456)# deque([456, 123, 333])print(q)# 打印：456print(q[0])q.pop()     # 从右边删除# deque([456, 123])print(q)q.popleft() # 从左边删除

队列四种模式

3）队列（生产者消费者模型）

1）生产者消费者模型的作用：

1、解决阻塞

2、就是解耦，修改生产者，不会影响消费者，反之亦然。

2）在生产环境，用生产者消费者模型，就可以解决：

1、处理瞬时并发的请求问题。瞬时的连接数就不会占满。所以服务器就不会挂了。

2、客户端提交一个请求，不用等待处理完毕，可以在页面上做别的事情。

2.1）如果不用队列存数据，服务端通过多线程来处理数据：

用户往队列存数据，服务器从队列里取数据。

没有队列的话，就跟最大连接数有关系，每个服务器就有最大连接数。

客户端要获取服务器放回，服务器要查、修改数据库或修改文件，要2分钟，那客户端就要挂起链接2分钟，2万个连接一半都要挂起，服务器就崩溃了。

如果没有队列，第一个用户发来请求，连上服务器，占用连接，等待2分钟。

第二个人来也要占用2分钟。

web服务器

如果要处理并发，有10万并发，如果：一台机器接收一个连接，需要10万个机器，等待2分钟就处理完了。

2.2）把请求放在队列的好处

用户发来请求，把请求放到队列里，可以让连接马上断开，不会阻塞，就不占用服务器的连接数了。如果看到订单处理了没，就要打开另外一个页面，查看请求是否处理。

服务器查询处理任务的时候，每个才花2分钟，服务器耗时是没有减少的。

但是这样做，客户端就不会持续的占用连接了。那瞬时的连接数就不会占满。所以服务器就不会挂了。

但是后台要处理10万个请求，也需要50台服务器。并不会减少服务器数量。

这样就能处理瞬时并发的请求问题。

服务器只是处理请求，是修改数据库的值，不是告诉客户端。而是客户端再发来请求，查询数据库已经修改的内容。

提交订单之后，把这个订单扔给队列，程序返回“正在处理”，就不等待了，然后断开这个连接，你可以在页面里做别的事情，不用一直等待订单处理完。这样就不影响服务器的最大连接数。在页面帮你发起一个alax请求，url，不停的请求（可能是定时器），我的订单成功没有，我的订单成功没有，如果订单成功了，就自动返回页面：订单成功

如果不用队列的话，一个请求就占用一个服务器，等待的人特别多，等待连接的个数太多了。服务器就挂掉了。

队列就没有最大个数限制，把请求发给队列了，然后http链接就断开了，就不用等待了。

12306买票的时候，下次再来请求的时候，就会告诉你，前面排了几个人。

3）python queue的特点：

python的queue是内存级别的。rabbitmq可以把队列发到别的服务器上处理。

所以python里的queue不能持久化，但是rabbitmq可以持久化。

queue.Queue()这样写，队列就没有最大个数限制。queue.Queue(5)就是说队列里最多能放5个值

4）生产者消费者代码示例：

import time,randomimport queue,threadingq = queue.Queue()def Producer(name):    count =0    while True:        time.sleep(random.randrange(3))        if q.qsize()<3:         # 只要盘子里小于3个包子，厨师就开始做包子            q.put(count)            print("Producer %s has produced %s baozi.." %(name,count))            count += 1def Consumer(name):    count =0    while True:        time.sleep(random.randrange(4))        if not q.empty():       # 只要盘子里有包子，顾客就要吃。            data = q.get()            print(data)            print('\033[32;1mConsumer %s has eat %s baozi...\033[0m' % (name,data))        else:           # 盘子里没有包子            print("---no baozi anymore----")        count+=1p1 = threading.Thread(target=Producer,args=('A',))c1 = threading.Thread(target=Consumer,args=('B',))c2 = threading.Thread(target=Consumer,args=('C',))p1.start()c1.start()c2.start()'''当你设计复杂程序的时候，就可以用生产者消费者模型，来松耦合你的代码,也可以减少阻塞。'''

实例代码

二. Python 多线程

1）线程锁

Lock，RLock

Lock只能锁一次，RLock可以递归多层，Lock不支持多层锁嵌套，我们一般用RLOCK

import threadingimport timeNUM = 10def func(lock):    global NUM    # 上锁    lock.acquire()    lock.acquire()    NUM -= 1    time.sleep(2)    print(NUM)    # 开锁    lock.release()    lock.release()# Lock = threading.Lock()   # 不支持嵌套锁，一般不用RLock = threading.RLock()   # 一般用RLock，支持嵌套锁。for i in range(10):    t = threading.Thread(target=func,args=(RLock,))    t.start()'''死锁：就是你也抢资源，我也抢资源，谁也抢不走就是死锁。如果是python，就是Lock，弄成嵌套锁，不支持，则变成死锁。解决办法：用RLock，支持嵌套锁'''

信号量 BoundedSemaphore

如果用线程锁，一次只允许一个进入，如果用信号量可以允许同时多少个一起进入。

每次5个线程同时执行，可能就会同时修改一个值。

import threadingimport timeNUM = 10def func(i,lock):    global NUM    # 上锁    lock.acquire()  # 总共30个 一次执行5个 25个，依次类推：20,15。。。    NUM -= 1    time.sleep(2)    print('NUM:',str(NUM),'i:',i)    # 开锁    lock.release()# Lock = threading.Lock()   # 不支持嵌套锁，一般不用# RLock = threading.RLock()   # 一般用RLock，支持嵌套锁。lock = threading.BoundedSemaphore(5)   # 参数是每次执行几个线程for i in range(30):    t = threading.Thread(target=func,args=(i,lock,))    t.start()

event红绿灯

要么全部阻塞（红灯），要么全部放开（绿灯）

import threadingdef func(i,e):    print(i)    # 10个线程并发打印：0-9 ，然后到wait的时候，就开始检测是什么灯    e.wait()    # 检测是什么灯，如果是红灯，停；绿灯，行    print(i+100)event = threading.Event()for i in range(10):    t = threading.Thread(target=func,args=(i,event))    t.start()event.clear()   # 设置成红灯inp = input('>>>')if inp == "1":    event.set() #设置成绿灯

线程锁条件-condition1

Lock,RLock：线程锁使用场景：

Lock,RLock是多个用户同时修改一份数据，可能会出现脏数据，数据就会乱，就加互斥锁，一次只能让一个人修改数据，就能解决。

condition，event，BoundedSemaphore 使用场景：

如果写了个爬虫，在建立数据库连接，线程就等着，什么能数据库能用了，就开通线程，再爬虫。

event是kua一下，全走了。

notify维护一个队列，传几个，就只能出去几次。

import threadingdef func(i, con):    print(i)    con.acquire()    con.wait()    print(i + 100)    con.release()c = threading.Condition()for i in range(10):    t = threading.Thread(target=func, args=(i, c,))    t.start()while True:    inp = input(">>>")    if inp == 'q':        break    c.acquire()    c.notify(int(inp))    c.release()

线程锁条件-condition2

con.wait_for里传一个函数名当参数，返回布尔值，是True，就执行下面的代码。反之，就不执行。

无论是否返回True，都是用了一个线程。

import threadingdef condition():    ret = False    r = input('>>>')    if r == 'true':        ret = True    else:        ret = False    return retdef func(i,con):    print(i)    con.acquire()    con.wait_for(condition)     # 只能一个一个过    print(i+100)    con.release()c = threading.Condition()for i in range(10):    t = threading.Thread(target=func, args=(i,c,))    t.start()

线程锁定时器

from threading import Timerdef hello():    print("hello, world")t = Timer(1, hello)  # 等1秒，执行hellot.start()  # after 1 seconds, "hello, world" will be printed

2)自定义线程池

自定义线程池基础版

import queueimport threadingimport timeclass TheadPool:    def __init__(self,maxsize = 5):        self.maxsize = maxsize        self._q = queue.Queue(maxsize)        for i in range(maxsize):    # 1、初始化的时候，先往队列里放5个线程            self._q.put(threading.Thread)        # 【threading.Thread, threading.Thread, threading.Thread, threading.Thread】    def get_thread(self):        return self._q.get()    def add_thread(self):        self._q.put(threading.Thread)pool = TheadPool(5)def task(arg,p):    # 2、线程并发执行，5个线程在瞬间（1秒钟之内）从队列里取出5个（执行get_thread()方法）    #    5个线程在瞬间打印0-4，就是i    print(arg)    time.sleep(1)   # 3、停了5秒    p.add_thread()  # 4、5个线程执行：队列添加线程（因为是5个线程执行，一个线程添加一个，队列总共是5个线程）for i in range(20): # 5、然后这样先取走5个，再put5个，然后打印i，就会出现：第一次打印：0-4，然后是：5-9,10-14,15-19    # threading.Thread类    t = pool.get_thread()    obj = t(target=task,args=(i,pool,))    obj.start()

这个程序的问题：

线程没有被重用，线程一下开到最大（浪费）

自定义线程池思路

不要把队列里放线程，而是放任务，开三个线程来从队列里取任务，如果都取完了，就会阻塞

方法1：

设置超时时间

方法2：

往队列尾部，加三个空值，如果取得是空值，则终止线程。

没有空闲线程，并且已经创建的线程小于最大的线程数，这样才会创建线程。

出现的问题

注意；之前的自定义线程池，如果定义queue的最多能放值的个数，pool = ThreadPool(5,5)

terminate就不好使了。

有的时候会一直阻塞，因为队列里已经有5个了，再往里面put一个，就超出queue里最大的个数。

自定义线程池代码

import queueimport threadingimport contextlibimport timeStopEvent = object()RUN = 0     # 定义线程池的三种状态CLOSE = 1TERMINATE = 2iNum=0'''开启最大个数为5个的队列，'''class ThreadPool(object):    def __init__(self, max_num, max_task_num = None):        if max_task_num:        # 如果传了最大队列数，就设置，否则就是无限大。            self.q = queue.Queue(max_task_num)        else:            self.q = queue.Queue()        self.max_num = max_num      # 设置最大线程数        self.cancel = False         # 假如已经执行close了，就不再执行任务，生成线程处理了        self.generate_list = []     # 已经生成的线程数列表        self.free_list = []         # 空闲的线程数列表        self._state = RUN    def run(self, func, args, callback=None):        """        线程池执行一个任务        :param func: 任务函数        :param args: 任务函数所需参数        :param callback: 任务执行失败或成功后执行的回调函数，回调函数有两个参数1、任务函数执行状态；2、任务函数返回值（默认为None，即：不执行回调函数）        :return: 如果线程池已经终止，则返回True否则None        """        if self.cancel:         # 假如已经执行close了，就不再执行任务，生成线程处理了            return        if len(self.free_list) == 0 and len(self.generate_list) < self.max_num: # 假如空闲的线程列表为空，并且已经生成的线程数小于最大线程数            self.generate_thread()      # 创建线程        w = (func, args, callback,)     # 把当前任务放入队列，也就是run循环了300次，就有300个任务放入队列        self.q.put(w)       # 注意：队列数是多少个，就要开启几个线程，因为当要关闭的线程池时，                            # 要把空对象加到队列。线程判断获取到是空对象（此时已经把queue里的任务都取完了）就关闭线程。        global iNum        iNum+=1        # print('qsize:',str(self.q.qsize()))    def generate_thread(self):        """        创建一个线程        """        t = threading.Thread(target=self.call)  # 执行call函数        t.start()    def call(self):        """        循环去获取任务函数并执行任务函数        """        current_thread = threading.currentThread    # 获取当前线程        self.generate_list.append(current_thread)   # 把当前线程加入到已经生成线程列表        event = self.q.get()        # 从队列里取一个任务        while event != StopEvent:   # 假如 这个任务不是空对象            func, arguments, callback = event      # 传进去的任务是个元组，由函数，参数，回调函数组成。            try:                result = func(*arguments)       # 执行任务，返回result                success = True                 # 执行成功，返回状态为True            except Exception as e:                success = False                result = None            else:                if callback is not None:       # 假如有回调函数                    try:                        callback(success, result)   # 把状态和返回值传给回调函数执行                    except Exception as e:                        pass            # 执行worker_state函数，空闲线程列表里是否加入个线程。在yield处执行with下的代码            with self.worker_state(self.free_list, current_thread):                if self._state == TERMINATE:        # 假如线程池状态是TERMINATE                    print(11111111111111111111111)                    event = StopEvent               # 就把当前任务赋值为空对象，while循环不满足，这样就走else的内容                else:                    event = self.q.get()            # 如果不是TERMINATE状态，则把当前任务赋给event对象        else:       # 如果while循环不满足，或者while循环完了，没有break，就执行else内容。            self.generate_list.remove(current_thread)   # 队列获取到了空对象，就关闭线程（从列表中移除当前的线程）            print(len(self.generate_list))    def close(self):        # 先执行close(),再执行join()        """        执行完所有的任务后，所有线程停止        """        if self._state == RUN:            self._state = CLOSE            self.cancel = True        full_size = len(self.generate_list)     # 查看已经生成的线程数个数        while full_size:            self.q.put(StopEvent)       # 往队列尾部加上一个空对象，由于队列是先进先出的，所以空对象是最后获取的，通过空对象就能关闭线程。            full_size -= 1      # 循环的次数为生成的线程的总个数    def terminate(self):        # 直接执行terminate()        """        无论是否还有任务，终止线程        """        self._state = TERMINATE        print("len:",str(len(self.generate_list)))        while self.generate_list:   # 假如线程列表不为空，就往队列里加上空对象            print('q.qsize():',str(self.q.qsize()))            self.q.get()            self.q.put(StopEvent)        # self.q = queue.Queue()        print(self.q.empty())       # 查看队列是否为空，相当于q.size==0        print('------------'+str(self.q.qsize()))    def join(self):     # CLOSE和join结合用        """Waits until all outstanding tasks have been completed."""        assert self._state in (CLOSE,)        delay = 0.0005        if self._state==CLOSE:            while self.q.qsize() > 0:                delay = min(delay * 2, .05)    @contextlib.contextmanager      # 上下文管理器    def worker_state(self, state_list, worker_thread):  # 传入的是空闲线程列表和当前线程        """        用于记录线程中正在等待的线程数        """        state_list.append(worker_thread)        # 把当前线程加到空闲线程里，yield前面的代码相当于执行__enter__,        try:            yield           # yield是执行with worker_state下的代码，        finally:            # yield后面的代码相当于执行__exit__            state_list.remove(worker_thread)    # 执行完一个queue的所有任务了，就移除这个线程了。因为一个队列对应着一个线程。pool = ThreadPool(5,5)def callback(status, result):    # status, execute action status    # result, execute action return value    passdef action(i):    print(i)for i in range(200):    ret = pool.run(action, (i,), callback)pool.terminate()# pool.close()# pool.join()print(1234234523452345234523452345234523452345234523455)# time.sleep(1)print(pool.q.qsize())print(len(pool.generate_list), len(pool.free_list))print('iNum:',iNum)# print(len(pool.generate_list), len(pool.free_list))

自定义线程池

三. Python 多进程

多进程

在windows里加main才能执行，如果在linux不加main可以执行。

在windows下，如果在程序里，不方便加main，只能放弃了。

from multiprocessing import Processdef foo(i):    print('say hi',i)if __name__ == '__main__':    for i in range(10):        p = Process(target=foo,args=(i,))        p.start()

1）进程之间的数据共享

这个例子是单线程，数据是共享的，无论单线程还是多线程都是共享的。

from multiprocessing import Processimport multiprocessingdef foo(i,li):    li.append(i)    print('say hi',i,li)if __name__ == '__main__':    li = []    for i in range(10):        # p = Process(target=foo,args=(i,li))        foo(i,li)        # p.daemon = True        # p.start()        # p.join()

进程默认数据不共享

#!/usr/bin/env python# -*- coding: utf-8 -*-__author__ = 'WangQiaomei'from multiprocessing import Processimport multiprocessingdef foo(i,li):    li.append(i)    print('say hi',i,li)if __name__ == '__main__':    li = []    for i in range(10):        p = Process(target=foo,args=(i,li))        # foo(i,li)        # p.daemon = True        p.start()        # p.join()

queues实现：进程之间数据共享

from multiprocessing import Processfrom multiprocessing import queuesimport multiprocessingdef foo(i,arg):    arg.put(i)    print('say hi',i,arg.qsize())if __name__ == '__main__':    # li =[]    li = queues.Queue(20,ctx=multiprocessing)    for i in range(10):        p = Process(target=foo,args=(i,li,))        # p.daemon = True        p.start()        # p.join()

数组和列表的区别：

数组和列表的特点比较：

1、数组类型一定：

数组只要定义好了，类型必须是一致的

python里列表里，可以放字符串也可以放数字。

2、数组个数一定：

创建数组的时候，就要指定数组多大，比如数组是10，再添加11个，就会报错

列表是动态的，个数不一定。

数组和列表的相邻元素的内存位置比较：

python的列表是基于c来实现。

python的列表相邻的两个元素在内存里，不一定挨着。是用链表实现的。

因为个数不限制，开始是10个长度，所以可能第11个被占用了。

每个元素，记录上一个和下一个的位置在哪里。可以找到位置。

字符串和int类型的占用内存的位置大小肯定不一样。所以数组，不只是长度一样，类型也要一样。

对于数组的话，相邻的元素是挨着的。

数组是int类型，并且长度是确定的。所以是相邻的。

数组的内存地址是连续的，列表不是，就是链表，链表是每个元素记录上一个位置和下一个位置在哪里。

数组实现：进程之间数据共享

from multiprocessing import Processfrom multiprocessing import queuesimport multiprocessingfrom multiprocessing import Arraydef foo(i,arg):    # arg.put(i)    # print('say hi',i,arg.qszie())    arg[i] = i + 100    for item in arg:        print(item)    print("====================")if __name__ == '__main__':    # li =[]    # li = queues.Queue(20,ctx=multiprocessing)    li = Array('i',5)    for i in range(5):        p = Process(target=foo,args=(i,li,))        # p.daemon = True        p.start()        # p.join()

注意：Array的参数，写了i，就只能放数字：

dict实现：进程之间数据共享

用Manager()对象创建一个特殊的字典。

For循环创建了多个进程，每个进程都可以利用dict。

dict.values()就是获取它所有的值，

如果字典获取的值是递增的，说明数据是共享的。

AttributeError:'ForkAwareLocal' object has no attribute 'connection'

如果把join注释就会报错：

报错：

conn = self._tls.connection

AttributeError: 'ForkAwareLocal' object has no attribute 'connection'

from multiprocessing import Processimport multiprocessingfrom multiprocessing import Managerdef foo(i,arg):    arg[i] = i + 100    print(arg.values())if __name__ == '__main__':    obj = Manager()    li = obj.dict()    for i in range(10):        p = Process(target=foo,args=(i,li,))        p.start()        # p.join()        # 方式1    # 方式2    # import time    # time.sleep(10)

原因是：

li = obj.dict()是在主进程创建的

for循环里创建的是子进程，子进程是修改主进程：arg[i] = i + 100（arg就是li）

主进程和子进程都在执行，主进程里有个字典，子进程要修改这个字典。

进程和进程之间要通信的话，需要创建连接的。相当于两边都写上一个socket，进程之间通过连接进行操作。

主进程执行到底部，说明执行完了，会把它里面的连接断开了。

主进程把连接断开了，子进程就连接不上主进程。

如果在底部写停10秒，主进程就停止下来，并没有执行完。主进程没有执行完，连接还没有断开，那子进程就可以连接它了。

解决方法1：停10秒（不建议）

from multiprocessing import Processimport multiprocessingfrom multiprocessing import Managerdef foo(i,arg):    arg[i] = i + 100    print(arg.values())if __name__ == '__main__':    obj = Manager()    li = obj.dict()    for i in range(10):        p = Process(target=foo,args=(i,li,))        p.start()        # p.join()        # 方式1    # 方式2    import time    time.sleep(10)

代码

测试：

如果把sleep(10)改成sleep(0.1),那么只会打印前面几行，然后又报之前的错误

打印：

[102]

[102, 105]

[102, 103, 105]

[102, 103, 105, 107]

接着报错：

conn = self._tls.connection

AttributeError: 'ForkAwareLocal' object has no attribute 'connection'

解决办法2：用join

p.join()

所以使用多进程的常规方法是，先调用start启动进程，再调用join要求主进程等待当前子进程的结束。

join是用来阻塞当前线程的，每次循环：p.start()之后，p就提示主线程，需要等待p结束才向下执行，那主线程就乖乖的等着啦。

from multiprocessing import Processimport multiprocessingfrom multiprocessing import Managerdef foo(i,arg):    arg[i] = i + 100    print(arg.values())if __name__ == '__main__':    obj = Manager()    li = obj.dict()    for i in range(10):        p = Process(target=foo,args=(i,li,))        p.start()        p.join()        # 方式1:    # 方式2    # import time    # time.sleep(10)

总结：进程之间共享的方式：

queues,数组和字典的方式

dict对类型没有限制，跟使用字典是一模一样的。用数组则限制了数据类型。

进程和进程之间要通信，是要连接的。

主进程执行到底部了，就执行完了，就把连接断开了。子进程就连不上主进程了。

2）进程锁

没有锁，多个进程就会一起修改数据：

加把锁就把进程锁住了，同一时间只有一个进程可以运行，其他都等着。

RLock, Lock, Event, Condition, Semaphore # 这些方法跟线程的使用方法是一样的

from multiprocessing import Processfrom multiprocessing import queuesfrom multiprocessing import Arrayfrom multiprocessing import RLock, Lock, Event, Condition, Semaphore    # 这些方法跟线程的使用方法是一样的import multiprocessingimport timedef foo(i,lis,lc):    lc.acquire()        # 加锁    lis[0] = lis[0] - 1    time.sleep(1)    print('say hi',lis[0])    lc.release()        # 释放锁if __name__ == "__main__":    # li = []    li = Array('i', 1)    li[0] = 10    lock = RLock()    for i in range(10):        p = Process(target=foo,args=(i,li,lock))        p.start()

3）进程池

进程池串行-apply

apply从进程池里取进程，然后一个一个执行，第一个进程执行完，第二个进程才执行，进程之间是串行的操作。这样就不是并发操作，没有太大意义。

from multiprocessing import Pooldef f1(arg):    print(arg)if __name__ == "__main__":    pool = Pool(5)    for i in range(10):        pool.apply(func=f1,args=(i,))   # apply执行函数，传入参数    print('end')

进程池异步-apply_rsync

from multiprocessing import Pooldef f1(arg):    print(arg)if __name__ == "__main__":    pool = Pool(5)    for i in range(10):        # pool.apply(func=f1,args=(i,))   # apply执行函数，传入参数        pool.apply_async(func=f1,args=(i,))    print('end')

这10个任务kua一下全执行了，主进程执行到end了。

主进程执行完了，子进程就被终止掉了。

主进程执行完了，就不再等子线程了，如果要等就要设置参数。

多线程线程默认也是，主进程不等子进程，多线程是：daemon=True加join来让他等。

主进程等子进程

1. close等子进程全部执行完

join是终止进程，必须要前面执行close或者terminate方法。

执行close，等所有任务（10个）全部执行完，再终止

执行terminate，表示立即终止，不管你当前的任务是否执行完，都立即终止。

from multiprocessing import Poolimport timedef f1(arg):    time.sleep(1)   # 加这句是为了看出5个5个执行的效果。    print(arg)if __name__ == "__main__":    pool = Pool(5)    for i in range(10):        # pool.apply(func=f1,args=(i,))   # apply执行函数，传入参数        pool.apply_async(func=f1,args=(i,))    pool.close()    pool.join()     # join表示：主进程执行到这里的时候，夯住了，等子进程结束的时候，再往下执行。    print('end')

光执行join，会触发下面的断言错误：

assert self._state in (CLOSE, TERMINATE)

join源代码有这句，只有符合这个条件的，才不会报错。
这个条件就是：执行join之前，必须执行close或者terminate方法。
close+join：是等子线程全部执行完了，才继续往下执行。

terminate立即终止

from multiprocessing import Poolimport timedef f1(arg):    time.sleep(1)    print(arg)if __name__ == "__main__":    pool = Pool(5)    for i in range(10):        # pool.apply(func=f1,args=(i,))   # apply执行函数，传入参数        pool.apply_async(func=f1,args=(i,))    time.sleep(1.5)    pool.terminate()   # 立即终止    pool.join()    print('end')

光执行join，会触发下面的断言错误：

assert self._state in (CLOSE, TERMINATE)

join源代码有这句，只有符合这个条件的，才不会报错。
这个条件就是：执行join之前，必须执行close或者terminate方法。
terminate+join：是表示立即终止，不管你当前的任务是否执行完，都立即终止。

四. 协程

1 协程及gevent原理

IO密集型：用多线程+gevent（更好），多线程

计算密集型：用多进程

协程原理：利用一个线程，分解一个线程成为多个“微线程”==>程序级别

如果写爬虫，就访问别的网站，拿别人源码。http请求叫IO请求，用多线程。

假设要访问3个url，创建3个线程，都在等待着，第一个有数据返回就继续执行，以此类推。

在等待过程中，就什么事也没干。

协程的方式。

计算机帮你创建进程、线程。线程是人为创建出来的。用一个线程，一会儿执行这个操作，一会儿执行那个操作。

协程是只用一个线程。程序员利用io多路复用的方式，让协程：

先访问一个url，不等待返回，就再访问第二个url，访问第三个url，然后也在等待。

greenlet本质是实现协程的。

注意：协程本身不高效，协程的本质只是程序员调用的，那为啥gevent这么高效率呢，是因为用了协程（greenlet）+IO多路复用的方式。

是IO多路复用的用法才能高效。所以用的时候就用gevent就好了。

用多线程：假设每爬一个网址需要2秒，3个url，就是3个请求，等待2秒，就可以继续往下走。

如果用gevent，用单线程，单线程应该从上到下执行，用for循环读取3个url，往地址发送url请求，就是IO请求，线程是不等待的。

for循环再拿第二个url，再发第三个url。在这过程中，谁先回来，就处理谁。

资源占用上，多线程占用了3个线程，2秒钟，多线程啥也没干，在等待。gevent在2秒钟，只要发送请求了，接着就想干什么干什么。

2. greenlet协程

from greenlet import greenletdef test1():    print(12)    gr2.switch()    print(34)    gr2.switch()def test2():    print(56)    gr1.switch()    print(78)gr1 = greenlet(test1)gr2 = greenlet(test2)gr1.switch()

3. gevent

gevent切换协程：

import geventdef foo():    print('Running in foo')         # 第1步    gevent.sleep(0)    print('Explicit context switch to foo again')   # 第3步def bar():    print('Explicit context to bar')    # 第2步    gevent.sleep(0)    print('Implicit context switch back to bar')    # 第4步gevent.joinall([    gevent.spawn(foo),    gevent.spawn(bar),])

4.gevent使用方法

from gevent import monkey; monkey.patch_all()import geventimport requests# 这个函数是发http请求的def f(url):    print('GET: %s' % url)    resp = requests.get(url)    data = resp.text    # 获取内容    print('%d bytes received from %s.' % (len(data), url))gevent.joinall([        gevent.spawn(f, 'https://www.python.org/'),     # 创建了一个协程        gevent.spawn(f, 'https://www.yahoo.com/'),      # 创建了一个协程        gevent.spawn(f, 'https://github.com/'),         # 创建了一个协程])

创建了三个协程。总共就一个线程，通过for循环发送三个url请求。然后等待结果，谁先回来，就处理谁。

通过requests.get(url)发送url请求，谁先回来，就拿到数据（data），拿到数据就可以处理数据了。

gevent的使用场景举例：

1、scrapy框架内部用的gevent。发请求性能比线程高很多。

2、做api（url）监控，把代码发布到哪个url，得自动检测下返回值是不是200，或是指定的状态码。

发布完成之后，就要发送http请求过去检测一下返回的状态码。如果有20个url请求，就用gevent一下全给发了，就没必要创建多个线程，一个线程就足以了，然后配合多进程+gevent，又可以利用多颗cpu的优势了。

monkey.patch_all()是什么？

发送http请求，是request本质上调用socket来发。原来执行http请求，就会通知我一下，执行完了，默认socket是没有这个功能的。这相当于把原来的socket修改了，修改成特殊功能的socket，发送请求如果完事了，会告诉你完事了。

其实内部就是把io请求做了个封装而已。

转载于:https://www.cnblogs.com/liuhailong-py-way/p/5680588.html

你可能感兴趣的文章

一些新的web性能优化技术