Python_多线程threading模块

时间：2022-02-08 06:05:11

python 在执行的时候会淡定的在CPU上只允许一个线程运行,故Python在多核CPU的情况下也只能发挥出单核的功能，其中的原因：gil锁

gil 锁 (全局解释器锁)：每个线程在执行时都需要先获取gil 一个线程运行Python，而其他N个睡眠或者等待I/O（即 保证同一时刻只有一个线程丢共享资源进行存取）

多线程两种调用方式：

import threading

import time

class Oh(threading.Thread):     # 继承

    # 多线程继承式调用

    def __init__(self,num):

        threading.Thread.__init__(self)   # （经典类写法）继承父类构造方法，否则会覆盖父类

        self.num = num

    def run(self):  # 定义每个线程要运行的函数

        print('我是一个数字：%s' % self.num)

        time.sleep(3)

if __name__ == '__main__':

    O1 = Oh(1)

    O2 = Oh(2)

    O1.start()

    O2.start()

继承式调用

import threading

import time

def Yes(num):     # 定义要运行的函数

    # 多线程直接式调用（常用）

    print('打印了一个数：%s'% num)

    time.sleep(3)                # 执行完等三秒

if __name__ =='__main__':

    y1 = threading.Thread(target=Yes, args=[1,])    # 实例化创建了一个线程

    y2 = threading.Thread(target=Yes, args=[2,])    # 第二个线程

    y1.start()     # 线程开始

    y2.start()

    print(y1.getName())   # 打印线程的名字

    print(y2.getName())

    y1.join()          #  主函数等待y1线程执行完过后再执行主线程

    y2.join()          # 线程为并行,全部执行完才一起等待 3 秒

    print('我是最后被打印的东东。。')        # 这是主线程执行的最后的东东

直接式调用

一、多线程方法

threading.enumerate()
 
返回当前运行中活着的线程对象列表
 
threading.active_count()
 
返回当前处于alive状态的线程对象个数（包含主线程），等于enumerate的列表长度
 
threading.current_thread() 
返回当前的线程对象，对应于调用者控制的线程
 
threading.get_ident() 
返回当前进程的‘线程标识符’
 
threading.main_thread()
 
返回主线程对象
 
threading.stack_size()  
返回当创建一个新线程使用的线程栈大小，0 为默认配置
 
threading.TIMEOUT_MAX 
设置threading全局超时时间
 
二、多线程类：
Thread 一个执行线程的对象
Lock 锁对象
RLock 可重入锁对象，使单一线程（再次）获得已持有的锁对象（递归锁）
Condition 条件变量对象，使得一个线程等待另外一个线程满足特定的条件，比如改变状态或者某个数据值
Event 条件变量的通用版本，任意数量的线程等待某个事件的发生，在该事件发生后所有的线程都将被激活       
Semaphore 为线程间的有限资源提供一个计数器，如果没有可用资源时会被阻塞
BoundedSemaphore 于Semaphore相似，不过它不允许超过初始值
Timer 于Thread类似，不过它要在运行前等待一定时间
Barrier 创建一个障碍，必须达到指定数量的线程后才可以继续

1.Thread类

Thread(group = None,target = None,name = None , args = (), kwargs = {})

  • group: 线程组，目前还没有实现，库引用中提示必须是None；
  • target: 要执行的方法；
  • name: 线程名；
  • args/kwargs: 要传入方法的参数

类中方法：

start() 启动线程，调用start(),run()
run() 定义线程的方法，经常被重写
join([timeout]) 阻塞到线程结束或到timeout值                      
getName() 获取线程名
setName() 设置线程名
is_alive() 返回线程是否正在运行
isDaemon() 是否是守护线程（已弃用）
setDaemon() 设置为守护线程，默认为Flase 


1>.start ()  & run()


    

  • start(): 启动一个子线程,调用start()和run()
    • 

  • run(): 只调用run()
    • 



2> 守护线程.setDaemon()


    

  • 默认为False，线程在前台运行，主线程执行过过程中，线程也在前台执行，主线程执行完毕后，等待线程执行完成，主线程再停止执行
    • 

  • 设置为True后，线程在后台运行，主线程执行过程中，线程也在后台执行，主线程执行完毕后，无论线程成功与否，完成与否均停止执行
    • 



例子：


1.循环等待最后一个线程（对join的操作）






import threading

import time

# 循环等待最后一个线程

def Vera(num):

    print('我是一个数：%s'% num)

    time.sleep(2)

t_list = []

if __name__ == '__main__':

    for i in range(10):

        v = threading.Thread(target=Vera,args=[i,])

        v.start()

        # v.join()    # 线程串行 # 主线程等待子线程 v 执行完

        t_list.append(v)         # 并行的线程，所有的都执行完

    for i in t_list:    # 所有的线程都执行join

        i.join()

    print('\n我完了完了完了。。')




example 1




2.守护线程和join中timeout的设置






import threading

import time

def run(n):

    print('第 【%s】个进程' % n)

    time.sleep(3)

    print('等待 【%s】秒后' % n)

join_list =[]

def main():

    for i in range(5):

        t = threading.Thread(target=run ,args=[i,])

        t.start()

        # t.join()    # 设置单线程走

        print('开始线程',t.getName())

        join_list.append(t)

    for n in join_list:    # 设置每一个线程都等待完（多线程（一起完，然后再执行sleep中的秒数））

        n.join()

m = threading.Thread(target=main,args =[])

m.setDaemon(True)    # 守护线程

m.start()

m.join(timeout=2)    # （子线程设为线程执行完全时）主线程被设为守护线程后 等待2秒后完

# （线程可以不执行完全时（即子线程内没有设置join））主线程被设为守护线程后，线程完全执行完但不等待sleep中的时间

print('主线程完了。。')




example 2




2.Lock & RLock类




gil是控制同一时刻在底层执行，是锁解释器级别以下的锁，只管锁底层而不管原生线程自己的内存数据间是否互斥，Lock则是加解释器以上的线程间的互斥，但是Python3.X中，貌似加了一层锁，但官方没有做相关解释，但为了保险起见，我们还是最好要加一层锁。


lock（指令锁）: 全局


RLock（可重入锁（递归锁））：线程


方法：


    

  • acquire([timeout]):尝试获得锁定，使线程进入同步阻塞状态
    • 

  • release():释放锁，使用前线程必须获得锁定，否则抛出异常
    • 







import threading

import time

def Presley():

    global num        # 获取num 全局变量

    print('我是一个数 ,我是[%s]'% num)

    time.sleep(1)       # 主要是打乱线程顺序

    lock.acquire()    # 锁线程

    num -=1

    lock.release()

lock = threading.Lock()

num = 10              # 共享变量

join_list = []

for i in range(10):       # 10个线程

    t = threading.Thread(target=Presley)

    t.start()

    join_list.append(t)

for t in join_list:      # 每个线程执行完全

    t.join()

print('我是最后的值over。。',num)




指令锁








import threading

import time 

def run1():   # 小锁一号

    print('我是run 1')

    lock.acquire()

    global num

    num +=1

    lock.release()

    return num

def run2():           # 小锁二号

    print('我是run 2')

    lock.acquire()

    global num2

    num2 +=1

    lock.release()

    return num2

def run3():         # 大锁 锁住了一号和二号

    print('我是run 3')

    lock.acquire()

    res = run1()       # 确保run 1 和run 2 中间没有其他的运行

    print('run 1 和 run 2')

    res2 = run2()

    lock.release()

    print(res, res2)

if __name__ == '__main__':

    num, num2 = 0, 0

    lock = threading.RLock()     # 每一个锁可以各自释放

    for i in range(10):

        t = threading.Thread(target=run3)

        t.start()

while threading.active_count() !=1:   # 当前还有几个线程

    print(threading.active_count())

else:

    print('所有的执行完了。。')

    print(num, num2)




递归锁




Lock & RLok（对比）






import threading

lock = threading.Lock() #Lock对象

lock.acquire()

lock.acquire()

print('yes')

lock.release()

lock.release()

print(lock.acquire())

# 结果

# 发生死锁，无线循环




Lock








import threading

rLock = threading.RLock()      #RLock对象

rLock.acquire()

rLock.acquire()

print('yes')

rLock.release()

rLock.release()

print(rlock.acquire())

# 结果

yes

True

#在同一线程内，程序不会堵塞。




RLock




3.Semaphore(信号量)&BoundSemaphore




互斥锁：同时只允许一个线程更改数据


Semaphore: 同时允许多个线程更改数据


Semaphore 在内部管理这一个计数器，调用 .acquire()时 ，计数器 -1 ，调用 .release()时，计数器 -1，而当计数器等于0时，acquire()则阻塞，直到其他线程来调用release()


BoundSemaphore 有界信号量会确保它当前值不超它的初始值，如果超过则抛出valueError异常


方法：


    

  • acquire():尝试获得锁定，使线程进入同步阻塞状态
    • 

  • release():释放锁
    • 







import threading,time

def run(n):

    se.acquire()

    time.sleep(1)

    print('运行线程：%s' % n)

    se.release()       # 释放，信号量 + 1

    # se.release()    # 再次释放，信号量 +1

    # 当指定为.Semaphore()时，多次的信号量 +1不会抛出异常

    # 当指定为.BoundeSemaphore()时，多次的信号量 +1 会抛出 ValueError 异常

if __name__ == '__main__':

    num = 0

    se = threading.Semaphore(3)

    # se = threading.BoundedSemaphore(3)

    for i in range(20):

        t = threading.Thread(target=run,args=[i,])

        t.start()

    print('主线程over..')

    print(num)




信号量




4.Event类




一个线程通知一个事件，另一个线程等待通知并作出处理


方法：


    

  • isSet(): 当内部标志为True则返回True，否则返回False
    • 

  • wait([timeout]): 不断检测set()是否阻塞，或者直到timeout超时
    • 

  • set(): 设置内部标志为True，所有等待的线程都被唤醒
    • 

  • clear(): 重新设置内部标志为False，调用wait()不断对set()检测直到set()被调用
    • 







import  threading,time

def light():

    if not event.isSet():

        event.set()        # 设置为Ture

    s = 0         # 假装这是设置的计时秒数

    while True:

        if s < 6:

            print('假装我是绿灯。。')

        elif s < 8:

            print('假装我是黄灯。。')

        elif s < 12:

            if event.isSet():

                event.clear()            # 红灯 所有车（线程）等待

            print('假装我是红灯。。')

        else:

            s = 0         # 秒数

            event.set()  # 重置为 绿灯

        time.sleep(2)

        s += 1        # 假设的秒数  加一

def car(n):

    while True:

        time.sleep(1)          # 每次的多个线程要一秒

        if event.isSet():

            print('车牌号【%s】,我过去啦。。啦。' % n)

        else:

            print('车牌号【%s】,我被塞住了。*_*。。。' % n)

            event.wait()    # 不断检测set()有没有被置为True

if __name__ == '__main__':

    event = threading.Event()

    Light = threading.Thread(target= light)

    Light.start()

    for i in range(5):     # 有 5 个车车在跑

        t = threading.Thread(target=car,args=[i, ])

        t.start()




车车过红绿灯的故事




未完待续。。
别人写的就是好（一份导图）： https://blog.csdn.net/fz420/article/details/78958745
相关文章