python之 多线程

Posted 2020-10-04

GIL全局解释器锁：

在Cpython 解释器中，同一个进程下开启的多线程，同一时刻只能有一个线程执行，无法利用多核优势。

所有的python代码都是交给解释器解释的，在同一进程中的多个线程以及解释器自带的垃圾回收线程是共享解释器资源的，共享就意味着竞争，竞争就会出现问题，比如说python线程想要执行一个一段代码，垃圾回收线程又想回收这段代码，这样就会出现错误。这时候必须有一种机制，保证数据安全，就是将这种并发的状态改为串行的状态。这种机制就是加锁处理，保证解释器同一时间只能执行一个任务的代码。

但是GIL只是保护解释器级别的数据安全，保护自己的数据还是需要自己加锁处理。

有了GIL锁，同一时刻同一进程中只有一个线程被执行！

 计算密集型：多线程效率高

from threading import Thread
from multiprocessing import Process
import time
def work():
    j = 0
    for i in range(10000000):
        j+= i
if __name__ == ‘__main__‘:
    L = []
    s = time.time()
    for i in range(4):
        # t = Thread(target = work)#10秒左右
        t = Process(target = work)#7秒左右
        L.append(t)
        t.start()
    for t  in L :
        t.join()
    stop = time.time()
    print("主 run time :%s"%(stop-s))

View Code

I/O密集型：多线程效率高

from multiprocessing import Process
from threading import Thread
def work():
    time.sleep(10)
if __name__ == ‘__main__‘:
    l = []
    start = time.time()
    for i in range(4):
        t = Thread(target = work)#主 run time:10.007068872451782
        # t = Process(target = work)#主 run time:12.057122707366943
        l.append(t)
        t.start()
    for t in l:
        t.join()
    stop = time.time()
    print("主 run time:%s"%(stop- start))

View Code

同步锁：

GIL锁是保护解释器级别的数据安全，那么程序自身的数据安全呢，这个时候就会用到同步锁。ps:同步锁就是互斥锁。锁通常是保护共享数据的安全性，当你需要访问该资源时，调用acquire方法来获取对象（如果其它线程已经获得该锁，则当前线程需要等待其释放）待资源访问完后，再调用release方法释放锁。

死锁与递归锁：

所谓死锁：是指两个或者两个以上的进程或线程在执行程序过程中，因争夺资源而造成的一种互相等待的现象，若无外力作用，会一直持续这种等待状态，这样的现象就是死锁。

from threading import Thread,RLock,Lock
import time
mutexA = Lock()
mutexB = Lock()
class Mythread(Thread):
    def run(self):
        self.f1()
        self.f2()
    def f1(self):
        mutexA.acquire()
        print("\033[33m%s 拿到 A锁"%self.name)
        mutexB.acquire()
        print("\033[34m%s 拿到 B锁"%self.name)
        mutexB.release()
        mutexA.release()
    def f2(self):
        mutexB.acquire()
        print("\033[35m%s 拿到 B锁"%self.name)
        time.sleep(1)
        mutexA.acquire()
        print("\033[36m%s 拿到 A锁"%self.name)
        mutexA.release()
        mutexB.release()
if __name__ == ‘__main__‘:
    for i in range(10):
        t = Mythread()
        t.start()
################################################
# Thread-1 拿到 A锁
# Thread-1 拿到 B锁
# Thread-1 拿到 B锁
# Thread-2 拿到 A锁

View Code

递归锁解决死锁问题：

在python中为了支持同一线程中的多次重复请求同一资源，python提供了可重入锁RLock。这个RLock内部维护着一个Lock和一个counter变量，counter记录了acquire的次数，从而使得资源可以多次require。直到一个线程所有的acquire都被release，其它线程才能获得资源。

from threading import Thread,RLock
mutexA = mutexB = RLock()
class Mythread(Thread):
    def run(self):
        self.f1()
        self.f2()
    def f1(self):
        mutexA.acquire()
        print(‘\033[43m%s 拿到 A 锁\033[0m‘%self.name)
        mutexB.acquire()
        print(‘\033[40m%s 拿到 B 锁\033[0m‘%self.name)
        mutexB.release()
        mutexA.release()
    def f2(self):
        mutexB.acquire()
        print(‘\033[41m%s 拿到 A 锁\033[0m‘ % self.name)
        time.sleep(1)
        mutexA.acquire()
        print(‘\033[42m%s 拿到 A 锁\033[0m‘ % self.name)
        mutexA.release()
        mutexB.release()
if __name__ == ‘__main__‘:
    for i in range(10):
        t = Mythread()
        t.start()

View Code

信号量Semaphore：

Semaphore 管理着内置的计数器，每当调用acquire（）时，内置计数器-1，调用release+1，当计数器为零时，acquire（）将阻塞线程直到其它线程调用release（）

这里的例子我们设置的最大连接数为5

from threading import Thread,Semaphore,currentThread
import time,random
sm = Semaphore(5)
def task():
    sm.acquire()
    print(‘\033[42m%s 正在上厕所‘%currentThread().getName())
    time.sleep(random.randint(1,3))
    print(‘\033[43m%s 上完厕所‘%currentThread().getName())
    sm.release()
if __name__ == ‘__main__‘:
    for i in range(10):
        t = Thread(target=task)
        t.start()

View Code

这里与进程池不同，进程池最大只能产生4个，而且从头到尾都只有这四个进程，不会产生新的，而信号量可以产生一堆。

Event（线程间通信）:

为了解决一个线程的运行需要依赖别的线程的结果这种现象，我们需要使用threading库中的Event对象，它包含一个可由线程设置的信号标志，它允许线程等待某些事件的发生。初始化时该标志位默认为False，如果有线程等待一个Event对象，而这个Event对象的标志为假，那么这个线程将会被一只阻塞直至该标志为真。一个线程如果将一个Event对象的标志位设置为真，那么它将唤醒所有等待这个Event对象的线程。

#红绿灯模拟

from threading import Thread,Event,currentThread
import time
e = Event()
def traffic_light():
    time.sleep(5)
    e.set()
def car():
    print("\033[40m%s is wait\033[0m"%(currentThread().getName()))
    e.wait()
    print("\033[44m%s is run\033[0m"%(currentThread().getName()))
if __name__ == ‘__main__‘:
    for i in range(10):
        t = Thread(target=car)
        t.start()
    traffic_thread = Thread(target=traffic_light)
    traffic_thread.start()

#模拟数据库请求
from threading import Thread,Event,currentThread
import time
e = Event()
def conn_mysql():
    count = 1
    while not e.is_set():
        if count>3:
            raise ConnectionError("尝试连接次数过多")
        print("\033[45m%s 第%s 尝试"%(currentThread().getName(),count))
        e.wait(timeout=1)
        count+= 1
    print("\033[44m%s 开始连接"%currentThread().getName())
def check_mysql():
    print("\033[40m%s 检测链接.....\033[0m "%currentThread().getName())
    time.sleep(3)
    e.set()
if __name__ == ‘__main__‘:
    for i in range(3):
        t = Thread(target=conn_mysql)
        t.start()
    t2 = Thread(target=check_mysql)
    t2.start()

View Code

定时器：

指定n秒后执行某操作

from threading import Timer


def hello(n):
    print("hello, world",n)


t = Timer(3, hello,args=(123,))
t.start()  # after 1 seconds, "hello, world" will be printed

View Code

线程queue：

import queue#先进先出
q = queue.Queue(3)
q.put("first")
q.put("second")
q.put("third")

print(q.get())
print(q.get())
print(q.get())

import queue#先进后出
q = queue.LifoQueue(3)
q.put(‘first‘)
q.put(‘second‘)
q.put(‘third‘)

print(q.get())
print(q.get())
print(q.get())
q = queue.PriorityQueue()#优先级
q.put((30,"first"))
q.put((10,"second"))
q.put((20,"third"))

print(q.get())
print(q.get())
print(q.get())

View Code