0820Python总结-线程队列,进程池和线程池,回调函数,协程

一.线程队列

from queue import Queue

put 存
get 取
put_nowait 存,超出了队列长度,报错
get_nowait 取,没数据时,直接报错
Linux Windows 线程中 put_nowait get_nowait 都支持

1.Queue

先进先出,后进后出

q = Queue()
q.put(1)
q.put(2)
print(q.get())
print(q.get())
# 取不出来,阻塞
# print(q.get())
# 没有数据时,报错
# print(q.get_nowait())

# 指定队列长度
q2 = Queue(3)
q2.put(100)
q2.put(101)
# q2.put(102)
# 存放的数据超出了队列长度,阻塞
# q2.put(103)
q2.put_nowait(104)

2.LifoQueue 先进后出,后进先出(栈的特点)

from queue import LifoQueue
lq = LifoQueue(3)
lq.put(11)
lq.put(22)
lq.put(33)
# lq.put_nowait(44)  errorprint(lq.get())
print(lq.get())
print(lq.get())
# print(lq.get())  阻塞

3.PriorityQueue 按照优先级进行排序(默认从小到大)

from queue import PriorityQueue
pq = PriorityQueue()
# 可以存放数字
# pq.put(80)
# pq.put(81)
# pq.put(18)# 可以存放字符串(按照ASCII编码进行排序,依次返回)
# pq.put("wangwen")
# pq.put("wangzhihe")
# pq.put("gelong")# 可以存放容器
# pq.put((18, "wangwen"))
# pq.put((18, "maohonglei"))
# pq.put((18, "wangawei"))# 是否可以将不同类型的数据都放到一个队列中的呢? 不可以
# error
"""
pq.put(1)
pq.put("abc")
"""print(pq.get())
print(pq.get())
print(pq.get())

二.进程池和线程池

from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
import os,time

def func(i):print("任务执行中...start", os.getpid())time.sleep(3)print("任务执行结束...end", i)return i

1.ProcessPoolExecutor 进程池的基本使用

if __name__ == '__main__':lst = []# cpu逻辑核心数print(os.cpu_count())# (1)创建进程池对象"""参数: 默认获取的是最大CPU逻辑核心数 8"""p = ProcessPoolExecutor(8)# (2)异步提交任务"""默认如果一个进程短时间内可以完成更多的任务,进程池就不会使用更多的进程来完成,以节省资源"""for i in range(10):res = p.submit(func, i)lst.append(res)# (3)获取当前进程任务中的返回值(result在获取任务的返回值时,有阻塞)for i in lst:print(i.result())# (4)等待所有子进程执行结束之后,再继续执行主进程内容(shutdown)# p.shutdown()  # <=> joinprint("=======")print(os.getpid())

2.ThreadPoolExecutor 线程池的基本使用

from threading import current_thread as cthreaddef func(i):print("thread...start", cthread().ident)print("thread...end", i)return cthread().identif __name__ == '__main__':lst = []setvar = set()# (1)创建线程池对象"""参数: 默认并发的线程数 是 os.cpu_count() * 5 = 40"""tp = ThreadPoolExecutor()# (2)异步提交任务"""默认如果一个线程短时间内可以完成更多的任务,线程池就不会使用更多的线程来完成,以节省资源"""for i in range(100):res = tp.submit(func, 10)lst.append(res)# (3)获取返回值for i in lst:setvar.add(i.result())# (4)等待所有子线程执行结束之后,再执行主线程# tp.shutdown()print("主线程执行结束...")print(setvar, len(setvar))

无论是进程池还是线程池,都是由固定的进程数或者线程数来执行所有的任务,并不额外创建多余的进程或者线程

3.线程池 map

from collections import Iterator,Iterable
def func(i):# 同一时间5个线程执行任务print("thread...", cthread().ident)return i * "*"if __name__ == '__main__':# 创建线程池对象(最大允许并发5个线程)tp = ThreadPoolExecutor(5)# 把执行的结果返回到迭代器中it = tp.map(func, range(20))# 判断返回值是迭代器print(isinstance(it, Iterator))# 等待所有子线程执行结束tp.shutdown()# 遍历迭代器for i in it:print(i)

三.回调函数

回调函数: 回头调用一下
把函数当成一个参数传递给另外一个函数
在当前函数执行完毕之后,最后调用一下当参数传递进来的函数
add_done_callback(回调函数)

功能:
支付状态
退款状态
转账的状态
把想要的相关的成员信息写在回调之后
通过支付接口调用之后,后台会自动把想要的数据加载到回调函数中
从而看到最后的状态

from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
from threading import current_thread as cthread
import os,time
def func1(i):print("process start...", os.getpid())time.sleep(1)print("process end...", i)return "*" * idef func2(i):print("thread start...", cthread().ident)time.sleep(1)print("thread end...", i)return "*" * idef call_back1(obj):print("===回调函数callback进程号===", os.getpid())print(obj.result())def call_back2(obj):print("===回调函数callback进程号===", cthread().ident)print(obj.result)

进程池的回调函数:由主进程执行调用完成的

if __name__ == '__main__':p = ProcessPoolExecutor()for i in range(1, 11):res = p.submit(func1, i)# print(res.result())res.add_done_callback(call_back1)# self.func(func2)p.shutdown()print("主进程执行结束...", os.getpid())

线程池的回调函数:由当前子进程调用完成的

if __name__ == '__main__':tp = ThreadPoolExecutor(5)for i in range(1, 11):res = tp.submit(func2, i)res.add_done_callback(call_back2)tp.shutdown()print("主线程执行结束...", cthread().ident)

"""
class Ceshi():def add_done_callback(self, func):print("执行操作1...")print("执行操作2...")func(self)  # call_back1234def result(self):return 1234123def call_back1234(obj):print(obj.result())obj = Ceshi()
res = obj.add_done_callback(call_back1234)
"""

四.协程

协程是线程的具体体现
安装协程模块 gevent

1.用协程改写生产者消费者模型

# 生产者
def producer():for i in range(100):yield i# 消费者
def consumer(gen):for i in range(10):print(next(gen))# 初始化生成器函数 -> 生成器
gen = producer()
consumer(gen)
consumer(gen)
consumer(gen)

2.协程的具体实现

from gevent import monkey
monkey.patch_all()
import time
import geventdef eat():print("eat 1")time.sleep(3)print("eat 2")def play():print("play 1")time.sleep(3)print("play 2")# 利用gevent.spawn创建协程对象g1
g1 = gevent.spawn(eat)
# 利用gevent.spawn创建协程对g2
g2 = gevent.spawn(play)# 如果不加join阻塞,默认主线程执行时,不等待直接结束
# 阻塞,必须等待g1协程任务执行完毕之后,放行
g1.join()
# 阻塞,必须等待g2协程任务执行完毕之后,放行
g2.join()print("主线程执行结束...")

五.协程相关方法

(1)spawn(函数,参数1,参数2…) 启动协程
(2)join 阻塞,直到某个协程任务执行完毕之后,再执行下面代码
(3)joinall 等待所有协程任务都执行完毕之后,放行
g1.join() g2.join() =>
gevent.joinall([g1,g2]) (推荐)
(4)value 获取协程任务中的返回值 g1.value g2.value

a = 1
b = 2
print(a, b)
# 通过分号把两句代码放到一行显示
a = 1;b = 2
print(a, b)

1.相应方法使用

from gevent import monkey;monkey.patch_all()
import time
import geventdef eat():print("eat 1")time.sleep(3)print("eat 2")return "吃完了"def play():print("play 1")time.sleep(3)print("play 2")return "玩完了"g1 = gevent.spawn(eat)
g2 = gevent.spawn(play)# 等待g1,g2协程任务执行完毕之后,再放行
gevent.joinall([g1, g2])print("主线程执行结束...")
print(g1.value)
print(g2.value)

2.利用协程爬取数据

HTTP 状态码:
200 OK
404 not food
400 bad request

import requests
response = requests.get("http://www.baidu.com/")
print(response)# 获取状态码
print(response.status_code)
# 获取网站中的编码
res = response.apparent_encoding
print(res)
# 设置编码集,防止乱码
response.encoding = res
# 获取网页里面的数据
res = response.text
print(res)

正常爬取

from gevent import monkey;monkey.patch_all()
import requests
import time
import geventurl_lst =[
"http://www.baidu.com/",
"http://www.taobao.com/",
"http://www.jingdong.com/",
"http://www.4399.com/",
"http://www.7k7k.com/",
]def get_url(url):response = requests.get(url)if response.status_code == 200:# print(response.text)pass
"""
startime = time.time()
for i in url_lst:get_url(i)
endtime = time.time()
print("执行时间:", endtime - startime)
"""# (2)用协程的方式爬取数据
lst = []
startime = time.time()
for i in url_lst:g = gevent.spawn(get_url, i)lst.append(g)gevent.joinall(lst)
endtime = time.time()
print("执行时间:", endtime - startime)

利用好多进程,多线程,多协程让服务器运行速度更快,抗住更大的并发