python并发编程之多线程（实践篇），,一.threadin

一.threading模块介绍

官网链接：https://docs.python.org/3/library/threading.html?highlight=threading#

1.开启线程的两种方式

#直接调用import threadingimport timedef run(n):    print(‘task‘,n)     time.sleep(2)t1 = threading.Thread(target=run,args=(‘t1‘,))t1.start()

#继承式调用mport threadingimport timeclass MyThread(threading.Thread):    def __init__(self,n,sleep_time):        super(MyThread, self).__init__()        self.n = n        self.sleep_time = sleep_time    def run(self):        print(‘running task‘,self.n)        time.sleep(self.sleep_time)        print(‘task done,‘,self.n)t1 = MyThread(‘t1‘,2)t1.start()

2.在一个进程下开启多个线程与在一个进程下开启多个子进程的区别

from threading import Threadfrom multiprocessing import Processimport osdef work():    print(‘hello‘)if __name__ == ‘__main__‘:    #在主进程下开启线程    t=Thread(target=work)    t.start()    print(‘主线程/主进程‘)    ‘‘‘    打印结果:    hello    主线程/主进程    ‘‘‘    #在主进程下开启子进程    t=Process(target=work)    t.start()    print(‘主线程/主进程‘)    ‘‘‘    打印结果:    主线程/主进程    hello    ‘‘‘

1.开启速度比较

from threading import Threadfrom multiprocessing import Processimport osdef work():    print(‘hello‘,os.getpid())if __name__ == ‘__main__‘:    #part1:在主进程下开启多个线程,每个线程都跟主进程的pid一样    t1=Thread(target=work)    t2=Thread(target=work)    t1.start()    t2.start()    print(‘主线程/主进程pid‘,os.getpid())    #part2:开多个进程,每个进程都有不同的pid    p1=Process(target=work)    p2=Process(target=work)    p1.start()    p2.start()    print(‘主线程/主进程pid‘,os.getpid())

2.比较pid

from  threading import Threadfrom multiprocessing import Processimport osdef work():    global n    n=0if __name__ == ‘__main__‘:    # n=100    # p=Process(target=work)    # p.start()    # p.join()    # print(‘主‘,n) #毫无疑问子进程p已经将自己的全局的n改成了0,但改的仅仅是它自己的,查看父进程的n仍然为100    n=1    t=Thread(target=work)    t.start()    t.join()    print(‘主‘,n) #查看结果为0,因为同一进程内的线程之间共享进程内的数据

3.数据是否共享

3.应用

1）将socket通信改写为多线程模式

#_*_coding:utf-8_*_#!/usr/bin/env pythonimport multiprocessingimport threadingimport sockets=socket.socket(socket.AF_INET,socket.SOCK_STREAM)s.bind((‘127.0.0.1‘,8080))s.listen(5)def action(conn):    while True:        data=conn.recv(1024)        print(data)        conn.send(data.upper())if __name__ == ‘__main__‘:    while True:        conn,addr=s.accept()        p=threading.Thread(target=action,args=(conn,))        p.start()

多线程并发的socket服务端

#_*_coding:utf-8_*_#!/usr/bin/env pythonimport sockets=socket.socket(socket.AF_INET,socket.SOCK_STREAM)s.connect((‘127.0.0.1‘,8080))while True:    msg=input(‘>>: ‘).strip()    if not msg:continue    s.send(msg.encode(‘utf-8‘))    data=s.recv(1024)    print(data)

客户端

2）三个任务，一个接收用户输入，一个将用户输入的内容格式化成大写，一个将格式化后的结果存入文件

rom threading import Threadmsg_l=[]format_l=[]def talk():    while True:        msg=input(‘>>: ‘).strip()        if not msg:continue        msg_l.append(msg)def format_msg():    while True:        if msg_l:            res=msg_l.pop()            format_l.append(res.upper())def save():    while True:        if format_l:            with open(‘db.txt‘,‘a‘,encoding=‘utf-8‘) as f:                res=format_l.pop()                f.write(‘%s\n‘ %res)if __name__ == ‘__main__‘:    t1=Thread(target=talk)    t2=Thread(target=format_msg)    t3=Thread(target=save)    t1.start()    t2.start()    t3.start()

View Code

3）主线程等待子线程结束

from threading import Threadimport timedef sayhi(name):    time.sleep(2)    print(‘%s say hello‘ %name)if __name__ == ‘__main__‘:    t=Thread(target=sayhi,args=(‘egon‘,))    t.start()    t.join()    #主线程等待子线程运行结束了再往下走    print(‘主线程‘)    print(t.is_alive())    ‘‘‘    egon say hello    主线程    False    ‘‘‘

join()方法

二.守护线程

无论是进程还是线程，都遵循：守护xxx会等待主xxx运行完毕后被销毁

1）对主进程来说，运行完毕指的是主进程代码运行完毕

2）对主线程来说，运行完毕指的是主线程所在的进程内所有非守护线程统统运行完毕，主线程才算运行完毕

需要强调的是：运行完毕并非终止运行

from threading import Threadimport timedef sayhi(name):    time.sleep(2)    print(‘%s say hello‘ %name)if __name__ == ‘__main__‘:    t=Thread(target=sayhi,args=(‘egon‘,))    t.setDaemon(True) #必须在t.start()之前设置    t.start()    print(‘主线程‘)    print(t.is_alive()) #结果为True说明此时主线程并没结束，守护进程还在    ‘‘‘    主线程    True    ‘‘‘

守护线程生命周期

from threading import Threadimport timedef foo():    print(123)    time.sleep(3)    print("end123")def bar():    print(456)    time.sleep(1)    print("end456")t1=Thread(target=foo)t2=Thread(target=bar)t1.daemon=True  #将t1设置为守护进程，主进程结束后t1也结束，t1.start()  #可能会出现t1没有完全完全走完就结束的情况t2.start()print("main-------")"""运行结果：123456main-------end456"""

案例分析

三.Python GIL(Global Interpreter Lock)

https://www.cnblogs.com/linhaifeng/articles/7449853.html

五.同步锁

1.GIL与lock

1）线程抢的是GIL锁，GIL锁相当于执行权限，拿到执行权限后才能拿到互斥锁Lock，其他线程也可以抢到GIL，但如果发现Lock仍然没有被释放则阻塞，即便是拿到执行权限GIL也要立刻交出来

2）join是等待所有，即整体串行，而锁只是锁住修改共享数据的部分，即部分串行，要想保证数据安全的根本原理在于让并发变成串行，join与互斥锁都可以实现，毫无疑问，互斥锁的部分串行效率要更高

3）GIL 与Lock是两把锁，保护的数据不一样，前者是解释器级别的（当然保护的就是解释器级别的数据，比如垃圾回收的数据），后者是保护用户自己开发的应用程序的数据，很明显GIL不负责这件事，只能用户自定义加锁处理，即Lock

2.过程分析

所有线程抢的是GIL锁，或者说所有线程抢的是执行权限

线程1抢到GIL锁，拿到执行权限，开始执行，然后加了一把Lock，还没有执行完毕，即线程1还未释放Lock，有可能线程2抢到GIL锁，开始执行，执行过程中发现Lock还没有被线程1释放，于是线程2进入阻塞，被夺走执行权限，有可能线程1拿到GIL，然后正常执行到释放Lock。。。这就导致了串行运行的效果

既然是串行，那我们执行

t1.start()

t1.join

t2.start()

t2.join()

这也是串行执行啊，为何还要加Lock呢，需知join是等待t1所有的代码执行完，相当于锁住了t1的所有代码，而Lock只是锁住一部分操作共享数据的代码。

3.Lock使用

锁通常被用来实现对共享资源的同步访问。为每一个共享资源创建一个Lock对象，当你需要访问该资源时，调用acquire方法来获取锁对象（如果其它线程已经获得了该锁，则当前线程需等待其被释放），待资源访问完后，再调用release方法释放锁：

import threadingR=threading.Lock()R.acquire()  #获取所对象‘‘‘对公共数据的操作‘‘‘R.release()  #释放

#1.100个线程去抢GIL锁，即抢执行权限#2. 肯定有一个线程先抢到GIL（暂且称为线程1），然后开始执行，一旦执行就会拿到lock.acquire()#3. 极有可能线程1还未运行完毕，就有另外一个线程2抢到GIL，然后开始运行，但线程2发现互斥锁lock还未被线程1释放，于是阻塞，被迫交出执行权限，即释放GIL#4.直到线程1重新抢到GIL，开始从上次暂停的位置继续执行，直到正常释放互斥锁lock，然后其他的线程再重复2 3 4的过程

GIL锁与互斥锁综合分析

#不加锁:并发执行,速度快,数据不安全from threading import current_thread,Thread,Lockimport os,timedef task():    global n    print(‘%s is running‘ %current_thread().getName())    temp=n    time.sleep(0.5)    n=temp-1if __name__ == ‘__main__‘:    n=100    lock=Lock()    threads=[]    start_time=time.time()    for i in range(100):        t=Thread(target=task)        threads.append(t)        t.start()    for t in threads:        t.join()    stop_time=time.time()    print(‘主:%s n:%s‘ %(stop_time-start_time,n))‘‘‘Thread-1 is runningThread-2 is running......Thread-100 is running主:0.5216062068939209 n:99‘‘‘#不加锁:未加锁部分并发执行,加锁部分串行执行,速度慢,数据安全from threading import current_thread,Thread,Lockimport os,timedef task():    #未加锁的代码并发运行    time.sleep(3)    print(‘%s start to run‘ %current_thread().getName())    global n    #加锁的代码串行运行    lock.acquire()    temp=n    time.sleep(0.5)    n=temp-1    lock.release()if __name__ == ‘__main__‘:    n=100    lock=Lock()    threads=[]    start_time=time.time()    for i in range(100):        t=Thread(target=task)        threads.append(t)        t.start()    for t in threads:        t.join()    stop_time=time.time()    print(‘主:%s n:%s‘ %(stop_time-start_time,n))‘‘‘Thread-1 is runningThread-2 is running......Thread-100 is running主:53.294203758239746 n:0‘‘‘#思考:既然加锁会让运行变成串行,那么我在start之后立即使用join,就不用加锁了啊,也是串行的效果啊#没错:在start之后立刻使用jion,肯定会将100个任务的执行变成串行,毫无疑问,最终n的结果也肯定是0,是安全的,但问题是#start后立即join:任务内的所有代码都是串行执行的,而加锁,只是加锁的部分即修改共享数据的部分是串行的#单从保证数据安全方面,二者都可以实现,但很明显是加锁的效率更高.from threading import current_thread,Thread,Lockimport os,timedef task():    time.sleep(3)    print(‘%s start to run‘ %current_thread().getName())    global n    temp=n    time.sleep(0.5)    n=temp-1if __name__ == ‘__main__‘:    n=100    lock=Lock()    start_time=time.time()    for i in range(100):        t=Thread(target=task)        t.start()        t.join()    stop_time=time.time()    print(‘主:%s n:%s‘ %(stop_time-start_time,n))‘‘‘Thread-1 start to runThread-2 start to run......Thread-100 start to run主:350.6937336921692 n:0 #耗时是多么的恐怖‘‘‘

互斥锁与join的区别

六.死锁现象与递归锁

所谓死锁： 是指两个或两个以上的进程或线程在执行过程中，因争夺资源而造成的一种互相等待的现象，若无外力作用，它们都将无法推进下去。此时称系统处于死锁状态或系统产生了死锁，这些永远在互相等待的进程称为死锁进程，如下就是死锁

from threading import Thread,Lockimport timemutexA=Lock()mutexB=Lock()class MyThread(Thread):    def run(self):        self.func1()        self.func2()    def func1(self):        mutexA.acquire()        print(‘\033[41m%s 拿到A锁\033[0m‘ %self.name)        mutexB.acquire()        print(‘\033[42m%s 拿到B锁\033[0m‘ %self.name)        mutexB.release()        mutexA.release()    def func2(self):        mutexB.acquire()        print(‘\033[43m%s 拿到B锁\033[0m‘ %self.name)        time.sleep(2)        mutexA.acquire()        print(‘\033[44m%s 拿到A锁\033[0m‘ %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锁然后就卡住，死锁了‘‘‘

死锁现象

解决方法，递归锁，在Python中为了支持在同一线程中多次请求同一资源，python提供了可重入锁RLock。

这个RLock内部维护着一个Lock和一个counter变量，counter记录了acquire的次数，从而使得资源可以被多次require。直到一个线程所有的acquire都被release，其他的线程才能获得资源。上面的例子如果使用RLock代替Lock，则不会发生死锁：

mutexA=mutexB=threading.RLock() #一个线程拿到锁，counter加1,该线程内又碰到加锁的情况，则counter继续加1，
#这期间所有其他线程都只能等待，等待该线程释放所有锁，即counter递减到0为止

七.信号量Semaphore

同进程的一样

Semaphore管理一个内置的计数器，
每当调用acquire()时内置计数器-1；
调用release() 时内置计数器+1；
计数器不能小于0；当计数器为0时，acquire()将阻塞线程直到其他线程调用release()。

实例：(同时只有5个线程可以获得semaphore,即可以限制最大连接数为5)：

from threading import Thread,Semaphoreimport threadingimport time# def func():#     if sm.acquire():#         print (threading.currentThread().getName() + ‘ get semaphore‘)#         time.sleep(2)#         sm.release()def func():    sm.acquire()    print(‘%s get sm‘ %threading.current_thread().getName())    time.sleep(3)    sm.release()if __name__ == ‘__main__‘:    sm=Semaphore(5)    for i in range(23):        t=Thread(target=func)        t.start()

View Code

与进程池是完全不同的概念，进程池Pool(4)，最大只能产生4个进程，而且从头到尾都只是这四个进程，不会产生新的，而信号量是产生一堆线程/进程

互斥锁与信号量推荐博客：http://url.cn/5DMsS9r

八.Event

同进程的一样

线程的一个关键特性是每个线程都是独立运行且状态不可预测。如果程序中的其 他线程需要通过判断某个线程的状态来确定自己下一步的操作,这时线程同步问题就会变得非常棘手。为了解决这些问题,我们需要使用threading库中的Event对象。 对象包含一个可由线程设置的信号标志,它允许线程等待某些事件的发生。在 初始情况下,Event对象中的信号标志被设置为假。如果有线程等待一个Event对象, 而这个Event对象的标志为假,那么这个线程将会被一直阻塞直至该标志为真。一个线程如果将一个Event对象的信号标志设置为真,它将唤醒所有等待这个Event对象的线程。如果一个线程等待一个已经被设置为真的Event对象,那么它将忽略这个事件, 继续执行。

event.isSet()：返回event的状态值；event.wait()：如果 event.isSet()==False将阻塞线程；event.set()： 设置event的状态值为True，所有阻塞池的线程激活进入就绪状态， 等待操作系统调度；event.clear()：恢复event的状态值为False。

import threading,timeevent = threading.Event()def lighter():    count = 0    event.set() #先设置绿灯    while True:        if count > 5 and count < 10:#改成红灯            event.clear()#标志位清了            print(‘\033[41;1mred light is on ...\033[0m‘)        elif count > 10:            event.set()#变绿灯            count = 0        else:            print(‘\033[42;1mgreen light is on ...\033[0m‘)        time.sleep(1)        count += 1def car(name):    while True:        if event.is_set():#代表绿灯            print(‘[%s] running...‘%name)            time.sleep(1)        else:            print(‘[%s] sees red light ,waiting ...‘ %name)            event.wait()            print(‘\033[34;1m[%s] green light is on,start going ... \033[0m‘ %name)light = threading.Thread(target=lighter,)light.start()car1 = threading.Thread(target=car,args=(‘宝马‘,))car1.start()

红绿灯

九.条件Condition

使得线程等待，只有满足某条件时，才释放n个线程

import threading def run(n):    con.acquire()    con.wait()    print("run the thread: %s" %n)    con.release() if __name__ == ‘__main__‘:     con = threading.Condition()    for i in range(10):        t = threading.Thread(target=run, args=(i,))        t.start()     while True:        inp = input(‘>>>‘)        if inp == ‘q‘:            break        con.acquire()        con.notify(int(inp))        con.release()

十.定时器

定时器，指定n秒后执行某操作

from threading import Timerdef hello():    print("hello, world") t = Timer(1, hello)t.start()  # after 1 seconds, "hello, world" will be printed

from threading import Timerimport random,timeclass Code:    def __init__(self):        self.make_cache()    def make_cache(self,interval=5):        self.cache=self.make_code()        print(self.cache)        self.t=Timer(interval,self.make_cache)        self.t.start()    def make_code(self,n=4):        res=‘‘        for i in range(n):            s1=str(random.randint(0,9))            s2=chr(random.randint(65,90))            res+=random.choice([s1,s2])        return res    def check(self):        while True:            inp=input(‘>>: ‘).strip()            if inp.upper() ==  self.cache:                print(‘验证成功‘,end=‘\n‘)                self.t.cancel()                breakif __name__ == ‘__main__‘:    obj=Code()    obj.check()

验证码定时器

十一.线程queue

queue队列 ：使用import queue，用法与进程Queue一样

classqueue.Queue(maxsize=0) #先进先出

import queueq=queue.Queue()q.put(‘first‘)q.put(‘second‘)q.put(‘third‘)print(q.get())print(q.get())print(q.get())‘‘‘结果(先进先出):firstsecondthird‘‘‘

View Code

classqueue.LifoQueue(maxsize=0) #last in fisrt out

import queueq=queue.LifoQueue()q.put(‘first‘)q.put(‘second‘)q.put(‘third‘)print(q.get())print(q.get())print(q.get())‘‘‘结果(后进先出):thirdsecondfirst‘‘‘

后进先出

classqueue.PriorityQueue(maxsize=0) #存储数据时可设置优先级的队列

import queueq=queue.PriorityQueue()#put进入一个元组,元组的第一个元素是优先级(通常是数字,也可以是非数字之间的比较),数字越小优先级越高q.put((20,‘a‘))q.put((10,‘b‘))q.put((30,‘c‘))print(q.get())print(q.get())print(q.get())‘‘‘结果(数字越小优先级越高,优先级高的优先出队):(10, ‘b‘)(20, ‘a‘)(30, ‘c‘)‘‘‘

设置优先级

python并发编程之多线程（实践篇）