线程队列,线程池,协程

Posted 2022-11-20 jie9527-

线程 queue

线程的queue，类似于进程

作用也是类似，queue(n)规范放入值的数量

queue.Queue(maxsize = 0)

这个和之前一样是为了实现先进先出

import queue
q = queue.Queue(2) # 括号内可加入数字规范放入值的数量，不加则不会规范
q.put('123')
q.put('qweqwe')
# q.put('111')
print(q.get())
print(q.get())
# print(q.get())
q.task_done()
q.join()

获得的结果是先get到‘123’

在这里如果加入多的q.put程序会阻塞，queue.Queue()则不会

class queue.LifoQueue(maxsize=0)

这则可以实现先进后出的效果

import queue
q = queue.LifoQueue() #堆栈 先进后出
q.put('蕾姆')
q.put('炭治郎')
q.put('绫乃')
print(q.get())
print(q.get())
print(q.get())

结果为：

绫乃
炭治郎
蕾姆

线程定时器

线程定时器可以设置默认几秒后开启一个线程，用t = Timer(n,线程名)来表示

from threading import Thread,Timer
import time


def task():
    print('线程执行了')
    time.sleep(2)
    print('线程结束了')


t = Timer(4,task) # 过了4s后开启了一个线程
t.start()
# print('hyc')

‘hyc’会在线程之前打印，而线程会在4秒后打印

进程池和线程池

进程池线程池:
池的功能限制进程数或线程数.
什么时候限制?
当并发的任务数量远远大于计算机所能承受的范围,即无法一次性开启过多的任务数量
就应该考虑去限制进程数或线程数,从保证服务器不崩.

from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
from threading import currentThread
from multiprocessing import current_process
import time

def task(i):
    print(f'currentThread().name 在执行任务 i')
    # print(f'进程 current_process().name 在执行任务 i')
    time.sleep(1)
    return i**2

if __name__ == '__main__':
    pool = ThreadPoolExecutor(4) # 池子里只有4个线程
    # pool = ProcessPoolExecutor(4) # 池子里只有4个线程
    fu_list = []
    for i in range(20):
        # pool.submit(task,i) # task任务要做20次,4个线程负责做这个事
        future = pool.submit(task,i) # task任务要做20次,4个进程负责做这个事
        # print(future.result()) # 如果没有结果一直等待拿到结果,导致了所有的任务都在串行
        fu_list.append(future)
    pool.shutdown() # 关闭了池的入口,会等待所有的任务执行完,结束阻塞.
    for fu in fu_list:
        print(fu.result())

同步异步

理解为提交任务的两种方式
同步: 提交了一个任务,必须等任务执行完了(拿到返回值),才能执行下一行代码,

异步: 提交了一个任务,不要等执行完了,可以直接执行下一行代码.

from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
from threading import currentThread
from multiprocessing import current_process
import time

def task(i):
    print(f'currentThread().name 在执行任务 i')
    # print(f'进程 current_process().name 在执行任务 i')
    time.sleep(1)
    return i**2


def parse(future):
    # 处理拿到的结果
    print(future.result())



if __name__ == '__main__':
    pool = ThreadPoolExecutor(4) # 池子里只有4个线程
    # pool = ProcessPoolExecutor(4) # 池子里只有4个线程
    fu_list = []
    for i in range(20):
        # pool.submit(task,i) # task任务要做20次,4个线程负责做这个事
        future = pool.submit(task,i) # task任务要做20次,4个进程负责做这个事
        future.add_done_callback(parse)
        # 为当前任务绑定了一个函数,在当前任务执行结束的时候会触发这个函数,
        # 会把future对象作为参数传给函数
        # 这个称之为回调函数,处理完了回来就调用这个函数.


        # print(future.result()) # 如果没有结果一直等待拿到结果,导致了所有的任务都在串行

    # pool.shutdown() # 关闭了池的入口,会等待所有的任务执行完,结束阻塞.
    # for fu in fu_list:
    #     print(fu.result())

协程

python的线程用的是操作系统原生的线程

协程:单线程下实现并发
并发:切换+保存状态
多线程:操作系统帮你实现的,如果遇到io切换,执行时间过长也会切换,实现一个雨露均沾的效果.

什么样的协程是有意义的?
遇到io切换的时候才有意义
具体:
协程概念本质是程序员抽象出来的,操作系统根本不知道协程存在,也就说来了一个线程我自己遇到io 我自己线程内部直接切到自己的别的任务上了,操作系统跟本发现不了,
也就是实现了单线程下效率最高.

优点:
自己控制切换要比操作系统切换快的多
缺点:
对比多线程
自己要检测所有的io,但凡有一个阻塞整体都跟着阻塞.
对比多进程
无法利用多核优势.
为什么要有协程(遇到io切换)?
自己控制切换要比操作系统切换快的多.降低了单个线程的io时间,

我们可以用协程来写一个没有io的代码

import time
def eat():
    print('eat 1')
    # 疯狂的计算呢没有io
    time.sleep(2)
    # for i in range(100000000):
    #     i+1
def play():
    print('play 1')
    # 疯狂的计算呢没有io
    time.sleep(3)
    # for i in range(100000000):
    #     i+1
play()
eat() # 5s

import time
def func1():
    while True:
        1000000+1
        yield

def func2():
    g = func1()
    for i in range(100000000):
        i+1
        next(g)

start = time.time()
func2()
stop = time.time()
print(stop - start)

它运行用了22秒的时间

如果我们不用协程去完成它会怎么样呢

import time

def func1():
    for i in range(100000000):
        i+1
def func2():
    for i in range(100000000):
        i+1

start = time.time()
func1()
func2()
stop = time.time()
print(stop - start) # 8.630893230438232

只用了8秒

可见，在没有io的情况下，不建议用协程。相反，有大量io的时候，协程就比较占优势了。而我们平时敲得代码都含有大量的io

gevent

我们也可以用gevent来实现协程，它相当于打了个补丁，可以实现捕获非gevent的io。当然，战士第三方库，必须得提前下载

from gevent import monkey;monkey.patch_all() #打了一个补丁,可以实现捕获非gevent的io.
import gevent

import time
def eat():
    print('eat 1')
    time.sleep(2)
    print('eat 2')
def play():
    print('play 1')
    # 疯狂的计算呢没有io
    time.sleep(3)
    print('play 2')

start = time.time()
g1 = gevent.spawn(eat)
g2 = gevent.spawn(play)
g1.join()
g2.join()
end = time.time()
print(end-start) # 3.0040290355682373