标签：默认 nal pool 生成 executor 同步 伪并发 join 导致
目录：
线程的三种状态:
1.就绪
2.运行
3.阻塞
阻塞和非阻塞描述的是运行的状态
阻塞 :遇到了IO操作,代码卡住,无法执行下一行,CPU会切换到其他任务
非阻塞 :与阻塞相反,代码正在执行(运行状态) 或处于就绪状态
同步和异步指的是提交任务的方式
同步 :提交任务必须等待任务完成,才能执行下一行
异步 :提交任务不需要等待任务完成,立即执行下一行
代码:
1 def task():
2 for i in range(1000000):
3 i += 1000
4 print("11111")
5
6 print("start")
7 task() # 同步提交方式,等函数运行完菜执行下一行
8 print("end")
9
10 from threading import Thread
11
12 print("start1")
13 Thread(target=task).start() # 异步提交,开启线程,然后去执行之后的代码,线程内代码自行执行
14 print("end1")
异步回调:
在发起异步任务后,子进程或子线程完成任务后需要通知任务发起方.通过调用一个函数,all_done_callback(函数名)
为什么需要回调? 子进程帮助主进程完成任务,处理任务的结果应该交还给主进程 其他方式也可以将数据交还给主进程 1.shutdown 主进程会等到所有任务完成 2.result函数 会阻塞直到任务完成 都会阻塞,导致效率降低,所以使用回调 注意: 回调函数什么时候被执行? 子进程任务完成时 谁在执行回调函数? 主进程 线程的异步回调 使用方式都相同,唯一的不同是执行回调函数,是子线程在执行(线程间数据共享)
1 # 方式1 自己来保存数据 并执行shutdown 仅在多线程
2
3 res = []
4 def task():
5 print("%s is 正在打水" % os.getpid())
6 time.sleep(0.2)
7 w = "%s 打的水" % os.getpid()
8 res.append(w)
9 return w
10
11 if __name__ == ‘__main__‘:
12 for i in range(20):
13 # 提交任务会返回一个对象 用于回去执行状态和结果
14 f = pool.submit(task)
15 print(f.result()) # 方式2 执行result 它是阻塞的直到任务完成 又变成串行了
16
17 print("11111")
18 # pool.shutdown() # 首先不允许提交新任务 然后等目前所有任务完成后
19 # print(res)
20 print("over")
21
22 ====================================================================================
23
24 pool = ThreadPoolExecutor()
25
26 # 方式3 通过回调(什么是回调 任务执行结束后自动调用某个函数)
27 def task():
28 print("%s is 正在打水" % os.getpid())
29 # time.sleep(0.2)
30 w = "%s 打的水" % os.getpid()
31 return w
32
33 def task_finish(res):
34 print("打水完成! %s" % res)
35
36 if __name__ == ‘__main__‘:
37 for i in range(20):
38 # 提交任务会返回一个对象 用于回去执行状态和结果
39 f = pool.submit(task)
40 f.add_done_callback(task_finish) #添加完成后的回调
41 print("11111")
42 print("over")
多进程:
1 from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
2 from threading import current_thread
3 import os
4 # 进程池
5 pool = ProcessPoolExecutor()
6 # 爬虫:从网络某个地址获取一个HTML文件
7 import requests # 该模块用于网络(HTTP)请求
8
9 # 生产数据,即生产者
10 def get_data_task(url):
11 print(os.getpid(),"正在生产数据!")
12 # print(current_thread(),"正在生产数据!")
13
14 response = requests.get(url)
15 text = response.content.decode("utf-8")
16 print(text)
17 return text
18
19 # 处理数据,即消费者
20 def parser_data(f):
21 print(os.getpid(),"处理数据")
22 # print(current_thread(), "处理数据")
23 print("正在解析: 长度%s" % len(f.result()))
24
25 urls = [
26 "http://www.baidu.com",
27 "http://www.baidu.com",
28 "http://www.baidu.com",
29 "http://www.baidu.com"
30 ]
31
32 if __name__ == ‘__main__‘:
33 for url in urls:
34 f = pool.submit(get_data_task,url)
35 f.add_done_callback(parser_data) # 回调函数是主进程在执行
36 # 因为子进程是负责获取数据的,然而数据怎么处理 ,子进程并不知道.应该把数据还给主进程
37 print("over")
多线程:
1 from concurrent.futures import ThreadPoolExecutor
2 from threading import current_thread
3 # 进程池
4 pool = ThreadPoolExecutor()
5
6 # 爬虫:从网络某个地址获取一个HTML文件
7 import requests # 该模块用于网络(HTTP)请求
8
9 # 生产数据
10 def get_data_task(url):
11 # print(os.getpid(),"正在生产数据!")
12 print(current_thread(),"正在生产数据!")
13
14 response = requests.get(url)
15 text = response.content.decode("utf-8")
16 print(text)
17 return text
18
19 # 处理数据
20 def parser_data(f):
21 # print(os.getpid(),"处理数据")
22 print(current_thread(), "处理数据")
23 print("正在解析: 长度%s" % len(f.result()))
24
25 urls = [
26 "http://www.baidu.com",
27 "http://www.baidu.com",
28 "http://www.baidu.com",
29 "http://www.baidu.com"
30 ]
31
32 if __name__ == ‘__main__‘:
33 for url in urls:
34 f = pool.submit(get_data_task,url)
35 f.add_done_callback(parser_data) # 回调函数是主进程在执行
36 # 因为子进程是负责获取数据的 然而数据怎么处理 子进程并不知道 应该把数据还给主进程
37 print("over")
普通队列/堆栈队列/优先级队列:
import queue
# 普通队列 先进先出
q = queue.Queue()
q.put("a")
q.put("b")
print(q.get())
print(q.get())
# 堆栈队列 先进后出 后进先出 函数调用就是进栈 函数结束就出栈 递归造成栈溢出
q2 = queue.LifoQueue()
q2.put("a")
q2.put("b")
print(q2.get())
# 优先级队列
q3 = queue.PriorityQueue() # 数值越小优先级越高 优先级相同时 比较大小 小的先取
q3.put((-100, "c"))
q3.put((1, "a"))
q3.put((100, "b"))
print(q3.get())
什么是协程？
协程指的是单线程下由应用程序级别实现的并发
即把本来由操作系统控制的切换+保存状态，在应用
程序里实现了
协程的切换vs操作系统的切换
优点：
切换速度远快于操作系统
缺点：
一个任务阻塞了，其余的任务都无法执行
ps：只有遇到io才切换到其他任务的协程才能提升
单线程的执行效率
为何用协程？
把单个线程的io降到最低，最大限度地提升单个线程的执行效率
如何实现协程？
from gevent import spawn,monkey;monkey.patch_all()
协程的目的是在单线程下实现并发
为什么出现协程? 因为cpython中,由于GIL而导致同一时间只有一个线程在跑
意味着:如果你的程序时计算密集,多线程效率也不会提升
如果是io密集型 没有必要在单线程下实现并发,我会开启多线程来处理io,子线遇到io,cpu切走.
不能保证一定切到主线
如果可以,我在遇到io的时候转而去做计算,这样一来可以保证cpu一直在处理你的程序,当然处理时间太长也要切走
总结:单线程下实现并发,是将io阻塞时间用于执行计算,可以提高效率
原理:一直使用CPU直到超时
怎么实现单线程并发?
并发:指的是看起来像是同时运行,实际是在任务间来回切换,同时需要保存执行的状态
任务一堆代码 可以用函数装起来
1.如何让两个函数切换执行
yield可以保存函数的执行状态
通过生成器可以实现伪并发
并发不一定提升效率,当任务全是计算时,反而会降低效率
2.如何知道发生了io, 从而切换执行?
第三方模块,gevent
第三方模块 greenlet 可以实现并发 但是不能检测io
第三方模块 gevent 封装greenlet 可以实现单线程并发,并且能够检测io操作,自动切换
协程的应用场景:
TCP 多客户端实现方式
1.来一个客户端就来一个进程 资源消耗较大
2.来一个客户端就来一个线程 也不能无限开
3.用进程池 或 线程池 还是一个线程或进程只能维护一个连接
4.协程 一个线程就可以处理多个客户端 遇到io就切到另一个
1 # 这是一个进程,默认包含一个主线程 2 import time
#生成器函数 3 def task(): 4 while True: 5 print("task1") 6 time.sleep(1)#I/O,CPU切走 7 yield 1 8 9 def task2(): 10 g = task() 11 while True: 12 try: 13 print("task2") 14 next(g)#next()函数参数传一个可迭代对象 15 except Exception: 16 print("任务完成") 17 break 18 task2() 19 打印结果: 20 task2 21 task1 22 task2 23 task1 24 task2 25 task1 26 ..........
# 1.实例化greenlet得到一个对象,传入要执行的任务,至少需要两个任务 # 2.先让某个任务执行起来,使用对象调用switch # 3.在任务的执行过程中,手动调用switch来切换
1 import greenlet
2 import time
3 def task1():
4 print("task1 1")
5 time.sleep(2)
6 g2.switch()
7 print("task1 2")
8 g2.switch()
9
10 def task2():
11 print("task2 1")
12 g1.switch()
13 print("task2 2")
14
15 g1 = greenlet.greenlet(task1)
16 g2 = greenlet.greenlet(task2)
17
18 g1.switch()
# 1.spawn函数传入你的任务 # 2.调用join 去开启任务 # 3.检测io操作需要打monkey补丁,就是一个函数,在程序最开始的地方调用它
1 from gevent import monkey 2 monkey.patch_all() 3 4 import gevent 5 import time 6 def eat(): 7 print(‘eat food 1‘) 8 time.sleep(2) 9 print(‘eat food 2‘) 10 11 def play(): 12 print(‘play 1‘) 13 time.sleep(1) 14 print(‘play 2‘) 15 16 g1=gevent.spawn(eat) 17 g2=gevent.spawn(play) 18 19 gevent.joinall([g1,g2]) 20 print(‘主‘)
服务端:
1 import gevent
2 from gevent import monkey
3 monkey.patch_all()
4 import socket
5
6 server = socket.socket()
7 # 重用端口
8 server.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1)
9
10 server.bind(("127.0.0.1",9999))
11
12 server.listen(5)
13 def data_handler(conn):
14 print("一个新连接..")
15 while True:
16 data = conn.recv(1024)
17 conn.send(data.upper())
18
19 while True:
20 conn,addr = server.accept()
21 # 切到处理数据的任务去执行
22 gevent.spawn(data_handler,conn)
标签：默认 nal pool 生成 executor 同步 伪并发 join 导致
原文地址：https://www.cnblogs.com/wyf2019/p/11047164.html
