因脉网站建设公司怎么呀韩国,怎么样优化关键词排名,中山市网站开发公司,婚庆网站名字文章目录0 前言1 并发、并行、同步、异步、阻塞、非阻塞1.1 并发1.2 并行1.3 同步1.4 异步1.5 阻塞1.6 非阻塞2 多线程2.1 Python线程的创建方式2.1.1 方式一2.1.2 方式二 继承Thread2.1.3 通过线程池创建多线程2.2 聊聊GIL2.2.1 Python线程与操作系统线程的关系2.3 线程同步2.…
文章目录0 前言1 并发、并行、同步、异步、阻塞、非阻塞1.1 并发1.2 并行1.3 同步1.4 异步1.5 阻塞1.6 非阻塞2 多线程2.1 Python线程的创建方式2.1.1 方式一2.1.2 方式二 继承Thread2.1.3 通过线程池创建多线程2.2 聊聊GIL2.2.1 Python线程与操作系统线程的关系2.3 线程同步2.3.1 加同步锁处理2.3.2 死锁问题2.3.3 可重入锁2.4 高级线程同步-Condtion2.5 高级线程同步-Semphore2.6 高级线程同步-Event3 IO 多路复用回调3.1 用同步的方式访问服务3.2 使用多线程提速3.3 改写成IO多路复用回调的方式4 协程4.1 yield关键字4.1.1 给yield传值4.1.2 给yield传异常4.1.3 gen.close() 关闭生成器4.2 yield from 关键字5 asyncio0 前言
高并发一直在软件开发遇到的老大难问题软件承载并发的能力也是一个核心性能点之一这篇文章主要讲解Python语言的高并发工具主要包括多进程、多线程、协程等。同时聊聊python的全局解释器锁对多线程的影响。
1 并发、并行、同步、异步、阻塞、非阻塞
1.1 并发 多任务在一个CPU上交替运行 通过CPU的时间片机制轮询调度多任务由于CPU执行速度非常快给人的感觉好像是多任务并行执行。
家里来客人了我们需要泡壶茶水招待客人 需要有以下几个工作要做 洗茶壶5分钟 洗茶杯3分钟 泡茶2分钟 烧水20分钟
主人这样做 先烧水烧水的同时洗茶壶和茶杯最后在泡茶总计耗时间为22分钟 主要忘记烧水了但是洗完茶杯又想起来了烧水的同时洗茶壶这样就需要消耗25分钟
其实这个生活小场景和并发非常相似烧水的过程中我们没有等水烧开再去做别的事情这就类似于操作系统的IO操作而是在烧水的过程中做一些其他的操作进而提高了工作效率。
1.2 并行
并行是多个任务同时执行真正的同时进行而不是通过CPU轮询执行。 举个生活中的例子一家人要吃晚饭爸爸妈妈孩子一起做
孩子负责洗菜 妈妈负责切菜 爸爸负责炒菜
三个人同时做不同的事情这就是并行。
1.3 同步
同步指的是不同任务同步进行一个任务完成之后等待结果返回后再开始另外一个任务。
import timedef task1():print(task1 start)time.sleep(1)print(task1 end)return task1 enddef task2():print(task2 start)time.sleep(2)print(task2 end)return task2 endif __name__ __main__:task1()task2()
执行结果task1和task2按顺序进行
task1 start
task1 end
task2 start
task2 end1.4 异步
异步指的是不同任务一起进行一个任务完成后不需要等待其返回结果直接进行另外一个任务。 异步编程在IO密集型场景中效率很高。
1.5 阻塞
阻塞的意思是函数等在某个位置直到达到某个条件后才往下走。
socket客户端的connect和recv都是阻塞点。
1.6 非阻塞
非阻塞是不必等待直接往下进行非阻塞一般会绑定回调函数进行工作经典的应用场景就是IO多路复用和回调函数。但是这种编程模型也有很多问题编程难度较高。
2 多线程
2.1 Python线程的创建方式
2.1.1 方式一
import time
import threadingdef task1():for i in range(10):time.sleep(1)print(一边看电视)def task2():for i in range(10):time.sleep(1)print(一边嗑瓜子)t1 threading.Thread(targettask1)
t2 threading.Thread(targettask2)t1.start()
t2.start()2.1.2 方式二 继承Thread
通过继承Thread类重写run方法实现多线程
import time
import threadingclass WatchThread(threading.Thread):def __init__(self, name):super().__init__()self.name namedef run(self):for i in range(10):time.sleep(1)print(一边看电视)class EatThread(threading.Thread):def __init__(self, name):super().__init__()self.name namedef run(self):for i in range(10):time.sleep(1)print(一边嗑瓜子)def main():wt WatchThread(name看电视线程)et EatThread(name嗑瓜子线程)wt.start()et.start()if __name__ __main__:main()
这两种实现多线程的方式中第二种是比较常用的第一种多用于测试或者一些初级应用场景
2.1.3 通过线程池创建多线程
通过python内置并发库里面的线程池开启多线程也是一种常用的方式线程池免去创建线程和销毁线程的开销对于那种频繁需要创建多线程和场景是比较合适的并且多线程之间的来回切换也会造成很大的系统级性能开销线程池维护指定数量的线程可以降低线程切换带来的性能开销同时可以控制线程的数量线程的数量并不是越多越好的。 map接口的例子 import time
from concurrent.futures import ThreadPoolExecutor, Executor, Futuredef task1(sleep_time):print(hello......)time.sleep(sleep_time)return worldexecutor ThreadPoolExecutor(max_workers3)fu executor.map(task1, [1,2,3,2,1]) # 同一个函数不同的参数print(fu) # generator object Executor.map.locals.result_iterator at 0x00000285EADDCD40for i in fu: # 获取结果print(i)submit结果的例子 import concurrent.futures
import urllib.requestURLS [http://www.foxnews.com/,http://www.cnn.com/,http://europe.wsj.com/,http://www.bbc.co.uk/,http://some-made-up-domain.com/]# Retrieve a single page and report the URL and contents
def load_url(url, timeout):with urllib.request.urlopen(url, timeouttimeout) as conn:return conn.read()# We can use a with statement to ensure threads are cleaned up promptly
with concurrent.futures.ThreadPoolExecutor(max_workers5) as executor:# Start the load operations and mark each future with its URLfuture_to_url {executor.submit(load_url, url, 60): url for url in URLS}for future in concurrent.futures.as_completed(future_to_url):url future_to_url[future]try:data future.result()except Exception as exc:print(%r generated an exception: %s % (url, exc))else:print(%r page is %d bytes % (url, len(data)))2.2 聊聊GIL
python的GIL一直被大家所诟病Python的执行速度确实很慢但是GIL不能完全背锅GIL只是不能发挥CPU的多核优势但是在非多线程环境下或者是IO密集型任务下跟GIL就没有关系了只是在计算密集型任务时会有一些比较慢的情况但是可以使用多进程加速程序的运行。 Python慢的本质原因是其所有的对象都被创建在堆内存中为了增加灵活性和动态性牺牲了运行速度这种设计理念也造成了Python的运行速度偏慢同时也因为Python的动态性和灵活性才能让他这么受欢迎有利有弊如果真的有项目语言运行速度成为瓶颈那可以更换其他编译型语言比如Go。
2.2.1 Python线程与操作系统线程的关系
Python通过内置的threading模块进行线程的创建threading模块是底层的_thread模块_thread模块是用C语言编写的编译之后内嵌到解释器中的。 所以Python创建线程对应C语言的线程C语言通过OS接口启动操作系统的线程所以Python的线程与操作系统的线程是一一对应的关系。 Python与CPython之间的关系 cpython是python语言的解释器cpython是c语言实现的python代码在执行前会被编译成字节码然后cpython解释器逐行执行字节码。 GIL跟python语言没有关系是cpython在实现的时候考虑到垃圾回收中的计数和C语言实现函数的原子性而设计的如果是Jpython解释器没有GIL。 重点GIL说的是Cpython解释器而不是python语言本身 GIL保证同一时刻只有一个Python线程能够执行字节码所以GIL是字节码级别的锁一条字节码对应一个或者多个C语言实现的函数GIL保证了C函数的原子性。 GIL并不能保证Python源码是线程安全性。
看下面这个程序
import disclass Ob:passobj Ob()def add():global objdel objdis.dis(add)
通过dis模块的dis方法获取add函数的字节码
其中第一列是源代码行号第二列是字节码偏移量第三列是操作指令也叫操作码第四列是指令参数也叫操作数。Python 的字节码指令都是成对出现的每个指令都会带有一个指令参数。
11 0 RESUME 013 2 DELETE_GLOBAL 0 (obj)4 LOAD_CONST 0 (None)6 RETURN_VALUE每个操作码都对应一个C函数
Python/bytecodes.c
inst(DELETE_GLOBAL, (--)) {PyObject *name GETITEM(names, oparg);int err;err PyDict_DelItem(GLOBALS(), name);// Cant use ERROR_IF here.if (err ! 0) {if (_PyErr_ExceptionMatches(tstate, PyExc_KeyError)) {format_exc_check_arg(tstate, PyExc_NameError,NAME_ERROR_MSG, name);}goto error;}}如果两个线程同时操作add函数 线程1在执行完err PyDict_DelItem(GLOBALS(), name);这句之后刚好发生了线程切换 线程2又会重新删除这个地址python中的对象对应C语言中的一个结构体这就造成了重复删除这个地址可能被其他程序使用了也可能是空的最终会造成什么问题没人知道。 为了保护C语言操作的原子性和引用计数所以才有了GIL这把大锁。
2.3 线程同步
刚说完GIL是字节码层面的锁保证cpython在同一时刻只能有一个线程执行字节码。但是python语言的层面还是会存在线程不安全的情况比如两个线程竞争同一个资源会造成数据不安全这就需要线程同步机制在python代码层面通过加锁来实现。 一个线程不安全的例子 from threading import Threadnum 0def add_num():global numfor i in range(1000000):num 1if __name__ __main__:t1 Thread(targetadd_num)t2 Thread(targetadd_num)t3 Thread(targetadd_num)t1.start()t2.start()t3.start()print(num)每次得到的数据都不一样
23167532.3.1 加同步锁处理
from threading import Thread, Locknum 0
lock Lock()def add_num():global numglobal lockfor i in range(1000000):lock.acquire()num 1lock.release()if __name__ __main__:t1 Thread(targetadd_num)t2 Thread(targetadd_num)t3 Thread(targetadd_num)t1.start()t2.start()t3.start()t1.join()t2.join()t3.join()print(num)加锁之后都是3000000不管运行多少次
3000000很明显加锁之后的速度变慢了加锁和释放锁一定有性能消耗所以能不加锁尽量不要加锁。
不加锁的时间是消耗的时间0.14963150024414062 加锁的时间是消耗的时间是3.216205358505249 相差的时间将近30倍了。
2.3.2 死锁问题
import threadinglock threading.Lock()def task_a():lock.acquire()lock.acquire()print(死锁了。。。。)lock.release()lock.release()threading.Thread(targettask_a).start()观察控制点会发现代码没有执行也没有结束这就是死锁当线程没有释放锁又去取另一把锁的时候会造成死锁。
2.3.3 可重入锁
对于同一个线程允许同时获取多把锁但是一定要保证获取锁和释放锁的次数是相同的。
import threadingrlock threading.RLock()def task_a():global rlockrlock.acquire()rlock.acquire()print(相同线程可以同时获取多把锁但是一定要与释放次数相同)rlock.release()rlock.release()print(end)threading.Thread(targettask_a).start() # 同一个线程可以获取多次锁不会死锁
输出结果
相同线程可以同时获取多把锁但是一定要与释放次数相同
end2.4 高级线程同步-Condtion
主人与小爱同学对话
主人小爱同学 小爱在请问有什么需要帮您 主人今天天气怎么样? 小爱今天天气晴朗阳光明媚 主人小爱同学 小爱在 主人今天是星期几 小爱今天是星期三 。。。
这种对话通过两个线程来交互实现通过加锁很难实现因为python多线程的调度机制用的是操作系统的线程调度机制python没办法直接控制操作系统如何调度线程所以使用默认的调度机制通过加锁来实现交替运行是不能保证百分百实现的还有人想通过延时操作实现因为遇到延时操作线程都会发生切换但是在处理多线程问题的时候基本都不能通过时间处理这是一个经验总结。
我们创建两个线程通过condition来实现这种复杂的锁机制 Condition类是在threading内置模块下的一个类常用的 API有如下几个
acquire 内部维护了一把可重入锁获取可重入锁release 释放可重入锁wait 线程等待调度notify 通知python开始调度线程
import threadingclass Owner(threading.Thread):def __init__(self, name, cond: threading.Condition):super().__init__()self.name nameself._cond conddef run(self) - None:with self._cond:print(主人小爱同学)self._cond.notify()self._cond.wait()print(主人今天天气怎么样?)self._cond.notify()self._cond.wait()print(主人小爱同学)self._cond.notify()self._cond.wait()print(主人今天是星期几?)self._cond.notify()self._cond.wait()# 唤醒所有线程结束程序self._cond.notify_all()class XiaoAi(threading.Thread):def __init__(self, name, cond: threading.Condition):super().__init__()self.name nameself._cond conddef run(self) - None:with self._cond:self._cond.wait()print(小爱在请问有什么需要帮您)self._cond.notify()self._cond.wait()print(小爱今天天气晴朗阳光明媚)self._cond.notify()self._cond.wait()print(小爱在)self._cond.notify()self._cond.wait()print(小爱今天是星期三)self._cond.notify()def main():condition threading.Condition()owner_thread Owner(name主人, condcondition)xa_thread XiaoAi(name小爱同学, condcondition)xa_thread.start() # 注意这两个线程的启动顺序特别重要不能反了反了就错了owner_thread.start()owner_thread.join()xa_thread.join()if __name__ __main__:main()
2.5 高级线程同步-Semphore
控制线程并发的数量
import random
import threading
import timeclass Task(threading.Thread):def __init__(self, sem):super().__init__()self.sem semdef run(self):with self.sem:print(f{threading.current_thread().name}:task)time.sleep(random.randint(1, 3))def main():sem threading.Semaphore(3)for i in range(10):t Task(semsem)t.start()main()
2.6 高级线程同步-Event
Event与Condition类似看源码可以发现Event也是通过Condition实现的Event具有如下接口
event.isSet()返回event的状态值event.wait()如果 event.isSet()False将阻塞线程event.set() 设置event的状态值为True所有阻塞池的线程激活进入就绪状态 等待操作系统调度event.clear()恢复event的状态值为False。通过Event可以控制两个线程的调度机制进而实现高级别的线程同步。
from threading import Thread,Event
import timeeventEvent()def light():print(红灯正亮着)time.sleep(3)event.set() #绿灯亮def car(name):print(车%s正在等绿灯 %name)event.wait() #等灯绿 此时event为False,直到event.set()将其值设置为True,才会继续运行.print(车%s通行 %name)if __name__ __main__:# 红绿灯t1Thread(targetlight)t1.start()# 车for i in range(10):tThread(targetcar,args(i,))t.start()3 IO 多路复用回调
平常常提起的selelctpollepoll都是IO多路复用的底层实现IO多路复用从OS层面了用户态和内核态的管理通过通知和回调的方式避免程序进行等待。
为了本节的测试工作首先使用flask启动一个web服务 flask服务端代码 import timefrom flask import Flaskapp Flask(__name__)app.route(/sleep/int:sleep_second)
def index(sleep_second):time.sleep(sleep_second)return fsleeping {sleep_second}s......if __name__ __main__:app.run(host0.0.0.0) 客户端代码 import socketdef get_url():host 192.168.8.133client socket.socket(socket.AF_INET, socket.SOCK_STREAM)client.connect((host, 5000))client.send(fGET /sleep/2 HTTP/1.1\r\nHost:{host}\r\nConnection:close\r\n\r\n.encode(utf-8))response_byte bwhile True:response client.recv(1024)response_byte responseif not response:breakresponse response_byte.decode(utf-8).split(\r\n\r\n)[1]print(response)client.close()if __name__ __main__:get_url()
运行程序之后可以获取到返回值
D:\Envs\py_venvs\venv_py3.11\Scripts\python.exe D:\code\高并发测试代码\coders\get_url.py
sleeping 2s......3.1 用同步的方式访问服务
import socket
import timedef get_url(sleep_time):host 192.168.8.133client socket.socket(socket.AF_INET, socket.SOCK_STREAM)client.connect((host, 5000))client.send(fGET /sleep/{sleep_time} HTTP/1.1\r\nHost:{host}\r\nConnection:close\r\n\r\n.encode(utf-8))response_byte bwhile True:response client.recv(1024)response_byte responseif not response:breakresponse response_byte.decode(utf-8).split(\r\n\r\n)[1]print(response)client.close()if __name__ __main__:start_time time.time()for i in range(10):get_url(i)print(f消耗的时间:{time.time()-start_time})
运行结果
sleeping 0s......
sleeping 1s......
sleeping 2s......
sleeping 3s......
sleeping 4s......
sleeping 5s......
sleeping 6s......
sleeping 7s......
sleeping 8s......
sleeping 9s......
消耗的时间:45.036475896835333.2 使用多线程提速
import socket
import threading
import timedef get_url(sleep_time):host 192.168.8.133client socket.socket(socket.AF_INET, socket.SOCK_STREAM)client.connect((host, 5000))client.send(fGET /sleep/{sleep_time} HTTP/1.1\r\nHost:{host}\r\nConnection:close\r\n\r\n.encode(utf-8))response_byte bwhile True:response client.recv(1024)response_byte responseif not response:breakresponse response_byte.decode(utf-8).split(\r\n\r\n)[1]print(response)client.close()if __name__ __main__:start_time time.time()for i in range(10):threading.Thread(targetget_url, args(i,)).start()while len(threading.enumerate())-1:passprint(f消耗的时间:{time.time()-start_time}) 运行结果 sleeping 0s......
sleeping 1s......
sleeping 2s......
sleeping 3s......
sleeping 4s......
sleeping 5s......
sleeping 6s......
sleeping 7s......
sleeping 8s......
sleeping 9s......
消耗的时间:9.0927593708038333.3 改写成IO多路复用回调的方式
import time
import socket
from selectors import DefaultSelector, EVENT_READ, EVENT_WRITEselect DefaultSelector()
urls []
stop Falseclass GetSource:def __init__(self, _url):self.host 192.168.8.133self.port 5000self.client socket.socket(socket.AF_INET, socket.SOCK_STREAM)self.client.setblocking(False)self.url _urlself.response_byte bdef get_connect(self):try:self.client.connect((self.host, self.port))except OSError:passselect.register(self.client.fileno(), EVENT_WRITE, self.send_msg)def send_msg(self, key):select.unregister(key.fd)self.client.send(fGET {self.url} HTTP/1.1\r\nHost:{self.host}\r\nConnection:close\r\n\r\n.encode(utf-8))select.register(self.client.fileno(), EVENT_READ, self.recv_msg)def recv_msg(self, key):response self.client.recv(1024)if response:self.response_byte responseelse:select.unregister(key.fd)response self.response_byte.decode(utf-8).split(\r\n\r\n)[1]print(response)self.client.close()urls.remove(self.url)if not urls:global stopstop Truedef loop():global stopwhile not stop:ready select.select()for key, mask in ready:call_back key.datacall_back(key)if __name__ __main__:start time.time()for i in range(0, 10):url f/sleep/{i}urls.append(url)gs GetSource(url)gs.get_connect()loop()print(f消耗的时间{time.time() - start})运行结果 sleeping 0s......
sleeping 1s......
sleeping 2s......
sleeping 3s......
sleeping 4s......
sleeping 5s......
sleeping 6s......
sleeping 7s......
sleeping 8s......
sleeping 9s......
消耗的时间9.008429527282715IO多路复用比多线程快10倍。
4 协程
协程的概念很早就被提出了大家都知道进程的切换会消耗很多系统资源,所以为了提高系统的并发性又出现了多线程操作系统在创建多线程的时候会在用户态和内核态分别创建内存空间供线程使用线程在执行过程中存在内核态和用户态的切换这个过程同样是需要消耗时间的。为了降低多线程切换的性能开销和用户态内核态的相互拷贝前辈们提出了协程的概念协程其实是用户态的多线程协程的调度完全由程序员负责操作系统是完全感觉不到的。协程的关键是函数能够在指定的位置停止并且能够恢复到停止之前的状态继续执行。
4.1 yield关键字
在函数中使用yield关键字调用函数之后返回一个生成器对象。
def gen_func():yield 1yield 2gen gen_func()print(gen) # generator object gen_func at 0x000001EFE4B9B1C0生成器函数的生命周期 GEN_CREATED 生成器被创建GEN_RUNNING 生成器运行中GEN_SUSPENDED 生成器运行中GEN_CLOSED 生成器已停止
def gen_func():yield 1yield 2gen gen_func()print(gen) # generator object gen_func at 0x000001EFE4B9B1C0import inspectprint(inspect.getgeneratorstate(gen)) # GEN_CREATEDres gen.send(None)
print(res)
print(inspect.getgeneratorstate((gen))) # GEN_SUSPENDEDres gen.send(None)
print(res)try:res gen.send(None)print(res)
except StopIteration:passprint(inspect.getgeneratorstate(gen)) # GEN_CLOSED
当函数定义中包括yield关键字后这就是一个生成器函数生成器函数调用后会返回一个生成器对象生成器对象必须驱动才能够运行驱动生成器有两种方式
gen_object.send(None)next(gen_object)
上面这两种方式是相同的作用生成器驱动之后第一次调用会停留在第一个yield之后并把yield后面的值返回给调用方第二次调用会停在第二个yield后面生成器函数执行结束后会抛出一个StopIteration异常。
同样生成器也是一个特殊的迭代器可以通过for语句进行遍历
def gen_func():yield 1yield 2for i in gen:print(i) 输出结果 1
2其实for语言的本质是
while Truetryval next(gen)print(val)except StopIteration:break4.1.1 给yield传值
调用方不仅能够收到yield出来的值还能够将值传递给生成器函数。
def gen_func():a yield 1print(f接收到的第一个值 a{a})b yield 2print(f接收到的第二个值 b: {b})return gen returngen gen_func()# 激活生成器对象
res gen.send(None)
# 获取第一个yield后面的值
print(res)
# 给生成器对象传递值同时获取第二个yield后面的值
res gen.send(hello)
# 生成器函数向下执行
print(res)
try:res gen.send(world)
except StopIteration as e:print(e.value) # gen return 执行结果 1
接收到的第一个值 ahello
2
接收到的第二个值 b: world
gen return4.1.2 给yield传异常
python中一切皆对象既然能够传值当然也能够传递异常。
def gen_func():a Nonetry:a yield 1except ValueError: # 如果不处理直接报错passprint(f接收到的第一个值 a{a})b yield 2print(f接收到的第二个值 b: {b})return gen returngen gen_func()# 激活生成器对象
res gen.send(None)
# 获取第一个yield后面的值
print(res)
# 给生成器对象传递值同时获取第二个yield后面的值
res gen.throw(ValueError(值传递错误))4.1.3 gen.close() 关闭生成器
def gen_func():yield 1yield 2gen gen_func()import inspectprint(inspect.getgeneratorstate(gen)) # GEN_CREATEDgen.send(None)
gen.close()
print(inspect.getgeneratorstate(gen)) # GEN_CLOSED
4.2 yield from 关键字
在一个生成器中yield另外一个生成器
def gen1():yield 1yield 2def gen2():yield gen1()gen2 gen2()
print(gen2) # generator object gen2 at 0x00000185E6115180res gen2.send(None)
for i in res:print(i)输出结果 generator object gen2 at 0x00000185E6115180
1
2使用yield from 关键字更简单的实现这个功能 def gen1():yield 1yield 2def gen2():yield from gen1()gen2 gen2()
print(gen2) # generator object gen2 at 0x00000185E6115180for i in gen2:print(i)直接遍历父生成器就可以获取到子生成器中的值 generator object gen2 at 0x000001606D2D5180
1
2yield from 可以作为委派生成器在调用方和子生成器之间搭建一个桥梁让调用方可以获取到子生成器的数据同时可以将数据或者异常直接通过委派生成器传送到子生成器中yield from也是python实现原生协程的根本所在。
5 asyncio
