前往小程序,Get更优阅读体验!
立即前往
首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >python3--线程,锁,同步锁,递归锁,信号量,事件,条件和定时器,队列,线程池

python3--线程,锁,同步锁,递归锁,信号量,事件,条件和定时器,队列,线程池

作者头像
py3study
发布2018-08-02 16:49:41
3K0
发布2018-08-02 16:49:41
举报
文章被收录于专栏:python3

线程

什么是线程?

线程是cpu调度的最小单位

进程是资源分配的最小单位

进程和线程是什么关系?

  线程是在进程中的 一个执行单位

  多进程 本质上开启的这个进程里就有一个线程

  多线程 单纯的在当前进程中开启了多个线程

线程和进程的区别:

  线程的开启 销毁 任务切换的时间开销小

  在同一个进程中数据共享

  能实现并发,但不能脱离进程

  进程负责管理分配资源 线程负责执行代码

GIL锁 -- 全局解释器锁

同一时刻只能有一个线程访问CPU -- 线程锁

Cpython会受到GIL影响

python程序效率下降

高计算型 -- 多线程会导致程序的效率下降

高IO型的 -- 可以使用多线程

守护线程

守护线程和守护进程的区别

守护进程是等待主进程代码结束之后就结束

守护线程是等待主线程都结束之后才结束

主线程等待其他线程结束,才结束

Thread类的其他方法

代码语言:javascript
复制
Thread实例对象的方法
  isAlive(): 返回线程是否活动的。
  getName(): 返回线程名。
  setName(): 设置线程名。

threading模块提供的一些方法:
  threading.currentThread(): 返回当前的线程变量。
  threading.enumerate(): 返回一个包含正在运行的线程的list。正在运行指线程启动后、结束前,不包括启动前和终止
  后的线程。
  threading.activeCount(): 返回正在运行的线程数量,与len(threading.enumerate())有相同的结果。

代码示例

代码语言:javascript
复制
from threading import Thread
import time
def func():
    # 需要延迟一下,不然会先执行print,那么结果会不一样
    time.sleep(0.1)  
    print(123)

t = Thread(target=func)
t.start()
print(t.is_alive())  # 线程是否运行

执行结果

True

123

示例2

代码语言:javascript
复制
from threading import Thread, currentThread, enumerate
import time
def func():
    print('-->', currentThread())   # 打印线程名,线程号
    # 需要延迟一下,不然会先执行print,那么结果会不一样
    time.sleep(0.1)
    print(123)

t = Thread(target=func)
t.start()
print(t.is_alive())  # 线程是否运行
print(t.getName())  # 打印线程名
t.setName('t1')  # 设置线程名
print(t.getName())  # 打印线程名
print(currentThread())  # 打印线程名,线程号
print(enumerate())  # 打印线程列表

执行结果

-->

-->

True

Thread-1

t1

[

123

同步锁

既然有了GIL锁,为什么还有其他锁?

示例代码:

代码语言:javascript
复制
import time
from threading import Thread
def func():
    global n
    temp = n
    time.sleep(0.1)
    n = temp - 1

n = 100
t_lst = []
for i in range(100):
    t = Thread(target=func)
    t.start()
    t_lst.append(t)
for t in t_lst:
    t.join()
print(n)

执行结果

99

GIL —— 全局解释器锁

锁线程 :在计算的时候 同一时刻只能有一个线程访问CPU

线程锁限制了你对CPU的使用,但是不影响web类或者爬虫类代码的效率

我们可以通过启动多进程的形式来弥补这个问题

为什么是99呢?

每个线程,从进程中取值100,那么CPU计算100-1,结果是99

这个99又赋值给n,进程变量就是99,所以每次都是赋值操作,赋值了100次,最终结果99,这样还是出现数据不安全的情况

如何解决?加锁

错误 : 示例代码如下

代码语言:javascript
复制
import time
from threading import Thread,Lock
def func(lock):
    global n
    lock.acquire()  # 加锁
    temp = n
    time.sleep(0.1)
    n = temp - 1
    lock.release()  # 释放锁(解锁)

n = 100
t_lst = []
for i in range(100):  # 写在for循环里面,等同于创建了100把锁
    lock = Lock()  # 创建对象锁
    t = Thread(target=func,args=(lock,))
    t.start()
    t_lst.append(t)
for t in t_lst:
    t.join()
print(n)

执行结果

99

正确代码示例:

代码语言:javascript
复制
import time
from threading import Thread,Lock
def func(lock):
    global n
    lock.acquire()
    temp = n
    time.sleep(0.1)
    n = temp - 1
    lock.release()

n = 100
t_lst = []
lock = Lock()  # 创建一个对象锁
for i in range(100):
    t = Thread(target=func,args=(lock,))
    t.start()
    t_lst.append(t)
for t in t_lst:
    t.join()
print(n)

执行结果

0

正确写法示例2

代码语言:javascript
复制
import time
from threading import Thread
def func():
    global n
    n = n -1

n = 100
t_lst = []
for i in range(100):
    t = Thread(target=func)
    t.start()
    t_lst.append(t)
for t in t_lst:
    t.join()
print(n)

执行结果,如果把计算和赋值两个步骤拆开,就会出现数据不安全的情况

0

总结:线程也需要锁,针对上面这张情况,需要加锁,这种锁,叫做同步锁

互斥锁

在同一个线程中,能够被一个锁的多个acquire阻塞住了,这种锁就叫互斥锁

代码语言:javascript
复制
from threading import Lock
lock = Lock()  # 在同一个线程中
               # 能够被一个锁的多个acquire阻塞住了
               # 这种锁就叫互斥锁
lock.acquire()
lock.acquire()

执行结果

阻塞住了......

死锁

进程也有死锁与递归锁

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

示例代码:

代码语言:javascript
复制
from threading import Lock
import time
mutexA = Lock()
mutexA.acquire()
mutexA.acquire()
time.sleep(0.1)
print(123)
mutexA.acquire()
mutexA.acquire()

执行结果

阻塞住......

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

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

示例代码:

代码语言:javascript
复制
from threading import RLock
import time
mutexA = RLock()
mutexA.acquire()
mutexA.acquire()
time.sleep(0.1)
print(123)
mutexA.acquire()
mutexA.acquire()

执行结果

123

递归锁 -- RLock

典型问题:科学家吃面

形成死锁原因--即需要同时满足两个必要因素,想要吃面--需要面和叉,造成资源的互相抢占问题

死锁代码示例:

代码语言:javascript
复制
import time
from threading import Thread,Lock
def eat1(noodle_lock,fork_lock,name):
    noodle_lock.acquire()
    print('%s抢到了面'%name)
    fork_lock.acquire()
    print('%s抢到了叉子'%name)
    print('%s正在吃面'%name)
    fork_lock.release()
    print('%s归还了叉子' % name)
    noodle_lock.release()
    print('%s归还了面' % name)

def eat2(noodle_lock,fork_lock,name):
    fork_lock.acquire()
    print('%s抢到了叉子' % name)
    time.sleep(0.5)
    noodle_lock.acquire()
    print('%s抢到了面'%name)
    print('%s正在吃面'%name)
    noodle_lock.release()
    print('%s归还了面' % name)
    fork_lock.release()
    print('%s归还了叉子' % name)

if __name__ == '__main__':
    noodle_lock = Lock()
    fork_lock = Lock()
    Thread(target=eat1,args=(noodle_lock,fork_lock,'zhangsan')).start()
    Thread(target=eat2,args=(noodle_lock,fork_lock,'lisi')).start()
    Thread(target=eat1,args=(noodle_lock,fork_lock,'wangwu')).start()
    Thread(target=eat2,args=(noodle_lock,fork_lock,'zhuliu')).start()

执行结果

blob.png
blob.png

只有一个锁,不会出现死锁,那么有多个锁的情况下,就会出现死锁

如何解决这个问题呢?使用递归锁

同一个线程中对同一个锁多次acquire不会产生阻塞

递归锁 -- 错误示例

代码语言:javascript
复制
from threading import Thread,RLock
def func(rlock,flag):
    rlock.acquire()
    print(flag*10)
    rlock.acquire()
    print(flag * 10)
    rlock.acquire()
    print(flag * 10)
    rlock.acquire()
    print(flag * 10)

rlock = RLock()
Thread(target=func,args=(rlock,'*')).start()
Thread(target=func,args=(rlock,'-')).start()

执行结果

blob.png
blob.png

出现死锁的情况

两个线程(进程),一个拿着外层钥匙,一个拿着内层钥匙,谁也不给对方,谁都进不去,就出现了死锁

blob.png
blob.png

注释:

第一个线程过来,拿走了一串钥匙

每次acquire,就进入一个房间,最后在房间的最里面,因为没有release,所以出不来,阻塞在里面

如图

blob.png
blob.png

它每走出一个房间,需要release一次,将钥匙放到最外面门上,让下个进程进去,所以有几次acquire,就有几次release,跟函数的递归类似,怎么解决上面卡住的问题?加上4次release就可以了,如下

代码语言:javascript
复制
from threading import Thread,RLock
def func(rlock,flag):
    rlock.acquire()
    print(flag*10)
    rlock.acquire()
    print(flag * 10)
    rlock.acquire()
    print(flag * 10)
    rlock.acquire()
    print(flag * 10)
    rlock.release()
    rlock.release()
    rlock.release()
    rlock.release()

rlock = RLock()
Thread(target=func,args=(rlock,'*')).start()
Thread(target=func,args=(rlock,'-')).start()

执行结果

**********

**********

**********

**********

----------

----------

----------

----------

使用递归锁,解决科学家吃面的问题,代码如下

代码语言:javascript
复制
import time
from threading import Thread,RLock
def eat1(noodle_lock,fork_lock,name):
    noodle_lock.acquire()
    print('%s抢到了面'%name)
    fork_lock.acquire()
    print('%s抢到了叉子'%name)
    print('%s正在吃面'%name)
    fork_lock.release()
    print('%s归还了叉子' % name)
    noodle_lock.release()
    print('%s归还了面' % name)

def eat2(noodle_lock,fork_lock,name):
    fork_lock.acquire()
    print('%s抢到了叉子' % name)
    time.sleep(0.5)
    noodle_lock.acquire()
    print('%s抢到了面'%name)
    print('%s正在吃面'%name)
    noodle_lock.release()
    print('%s归还了面' % name)
    fork_lock.release()
    print('%s归还了叉子' % name)

if __name__ == '__main__':
    fork_lock = noodle_lock = RLock()  # 表示同一串钥匙
    Thread(target=eat1,args=(noodle_lock,fork_lock,'zhangsan')).start()
    Thread(target=eat2,args=(noodle_lock,fork_lock,'lisi')).start()
    Thread(target=eat1,args=(noodle_lock,fork_lock,'wangwu')).start()
    Thread(target=eat2,args=(noodle_lock,fork_lock,'zhuliu')).start()

执行结果

download.gif
download.gif

什么情况下,需要用到递归锁呢?

有超过一个资源需要锁的时候 -- 递归锁

信号量

同进程一样

Semaphore管理一个内置的计数器,

每当调用acquire()时内置计数器-1;

调用release() 时内置计数器+1;

计数器不能小于0;当计数器为0时,acquire()将阻塞线程直到其他线程调用release()。

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

代码语言:javascript
复制
from threading import Thread,Semaphore,current_thread
import time
def func():
    sm.acquire()
    print('{} get sm'.format(current_thread().getName()))
    time.sleep(3)
    sm.release()

if __name__ == '__main__':
    sm = Semaphore(2)
    for i in range(5):
        t = Thread(target=func)
        t.start()

执行结果

Thread-1 get sm

Thread-2 get sm

Thread-3 get sm

Thread-4 get sm

Thread-5 get sm

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

事件

同进程一样

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

代码语言:javascript
复制
event.isSet():返回event的状态值;
event.wait():如果 event.isSet()==False将阻塞线程;
event.set(): 设置event的状态值为True,所有阻塞池的线程激活进入就绪状态, 等待操作系统调度;
event.clear():恢复event的状态值为False。
blob.png
blob.png

例如,有多个工作线程尝试链接MySQL,我们想要在链接前确保MySQL服务正常才让那些工作线程去连接MySQL服务器,如果连接不成功,都会去尝试重新连接。那么我们就可以采用threading.Event机制来协调各个工作线程的连接操作

实例:

代码语言:javascript
复制
from threading import Thread,Event,current_thread
import time
import random

def conn_mysql(event):
    count = 1
    while count < 4:
        print('第{}次尝试连接'.format(current_thread().getName(),count))
        event.wait(0.5)
        # 如果不传参数会一直等到事件为True为止
        # 如果传参数,传一个时间参数,等到多少秒,类似time.sleep()
        count += 1
        if event.is_set():
            print('连接成功'.format(current_thread().getName()))
            break
    else:
        print('连接失败')

def check_mysql(event):
    print('{}正在检查mysql'.format(current_thread().getName()))
    time.sleep(random.randint(1,2))
    event.set()


if __name__ == '__main__':
    event = Event()
    conn1 = Thread(target=conn_mysql,args=(event,))
    check = Thread(target=check_mysql,args=(event,))
    conn1.start()
    check.start()

执行结果

Thread-1第1次尝试连接

Thread-2正在检查mysql

Thread-1第2次尝试连接

Thread-1第3次尝试连接

Thread-1连接成功

条件

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

详细说明

代码语言:javascript
复制
Python提供的Condition对象提供了对复杂线程同步问题的支持。Condition被称为条件变量,除了提供与Lock类似的acquire
和release方法外,还提供了wait和notify方法。线程首先acquire一个条件变量,然后判断一些条件。如果条件不满足则
wait;如果条件满足,进行一些处理改变条件后,通过notify方法通知其他线程,其他处于wait状态的线程接到通知后会重
新判断条件。不断的重复这一过程,从而解决复杂的同步问题。

代码说明:

代码语言:javascript
复制
import threading
def run(n):
    con.acquire()
    con.wait()
    print("run the thread: {}".format(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('>>>').strip()
        if inp == 'q':
            break
        con.acquire()
        con.notify(int(inp))
        con.release()
        print('******')

执行结果

download.gif
download.gif

定时器

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

代码语言:javascript
复制
from threading import Timer
def hello():
    print('hello, world')

t = Timer(2, hello)  # 2秒后执行hello
t.start()

执行结果

hello, world

线程队列

queue队列 :使用import queue,用法与进程Queue一样

queue is especially useful in threaded programming when information must be exchanged safely between multiple threads.

class queue.Queue(maxsize=0) #先进先出

代码语言:javascript
复制
import queue
q = queue.Queue()  # 先进先出,先打印1
q.put('1')
q.put('2')
q.put('3')
print(q.get())
print(q.get())
print(q.get())

执行结果

1

2

3

class queue.LifoQueue(maxsize=0) #后进先出

代码语言:javascript
复制
import queue
q = queue.LifoQueue()  # 后进先出,先打印3
q.put('1')
q.put('2')
q.put('3')
print(q.get())
print(q.get())
print(q.get())

执行结果

3

2

1

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

优先级队列

代码语言:javascript
复制
import queue
q = queue.PriorityQueue()  # 优先级队列,数字越小优先级越高
                           # 字符串,则按照asicc码顺序排列
q.put((1,'a'))
q.put((10,'b'))
q.put((5,'c'))
print(q.get())
print(q.get())
print(q.get())

执行结果

(1, 'a')

(5, 'c')

(10, 'b')

Python标准模块--concurrent.futures

线程池

能够在多线程的基础上进一步节省内存和时间开销

代码语言:javascript
复制
#1 介绍
concurrent.futures模块提供了高度封装的异步调用接口
ThreadPoolExecutor:线程池,提供异步调用
ProcessPoolExecutor: 进程池,提供异步调用
Both implement the same interface, which is defined by the abstract Executor class.

#2 基本方法
#submit(fn, *args, **kwargs)
异步提交任务

#map(func, *iterables, timeout=None, chunksize=1) 
取代for循环submit的操作

#shutdown(wait=True) 
相当于进程池的pool.close()+pool.join()操作
wait=True,等待池内所有任务执行完毕回收完资源后才继续
wait=False,立即返回,并不会等待池内的任务执行完毕
但不管wait参数为何值,整个程序都会等到所有任务执行完毕
submit和map必须在shutdown之前

#result(timeout=None)
取得结果

#add_done_callback(fn)
回调函数

示例代码

代码语言:javascript
复制
import time
from concurrent.futures import ThreadPoolExecutor
def func(i):
    print(i*'*')
    time.sleep(1)

thread_pool =  ThreadPoolExecutor(5)  # ThreadPoolExecutor(5)设置线程数为5
for i in range(10):
    thread_pool.submit(func, i)  # submit 相当于apply_async 异步
thread_pool.shutdown()  # shutdown 相当于 close+join
print('1111')  # 如果想最后打印 1111,需要执行shutdown()

执行结果

*

**

***

****

*****

******

*******

********

*********

1111

上面代码 return一个值 异步代码示例

代码语言:javascript
复制
import time
from concurrent.futures import ThreadPoolExecutor
def func(i):
    print(i*'*')
    time.sleep(1)
    return i ** 2

thread_pool = ThreadPoolExecutor(5)  # 设置线程池数量为5
ret_lst = []
for i in range(10):
    ret = thread_pool.submit(func, i)  # submit 相当于apply_async 异步
    ret_lst.append(ret)
thread_pool.shutdown()  # shutdown 相当于 close + join
for ret in ret_lst:
    print(ret.result())  # result 返回值
print('111111')

执行结果

download.gif
download.gif

map的用法

代码语言:javascript
复制
from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
import os,time,random
def task(n):
    print('{} is runing'.format(os.getpid()))
    time.sleep(random.randint(1, 3))
    return n**2

if __name__ == '__main__':
    executor = ThreadPoolExecutor(max_workers=3)
    executor.map(task, range(1,12))  # map取代了for + submit

执行结果

download.gif
download.gif

回调函数

代码语言:javascript
复制
import time
from concurrent.futures import ThreadPoolExecutor
def func(i):
    print(i*'*')
    time.sleep(1)
    return i**2

def callb(arg):
    print(arg.result()*'-')

if __name__ == '__main__':
    thread_pool = ThreadPoolExecutor(5)
    for i in range(10):
        # 相当于apply_async  add_done_callback 回调函数
        thread_pool.submit(func,i).add_done_callback(callb)  #将callb函数的返回值作为参数
    thread_pool.shutdown()  # 相当于 close+join
    print('wahaha')

执行结果

download.gif
download.gif

当内存不需要共享,且高计算的时候 用进程

当内存需要共享,且高IO的时候 用线程

当并发很大的时候

    设计模式

    多进程 : 多个任务 —— 进程池 :cpu个数、cpu个数+1

    多线程 :多个任务  —— 线程池 :cpu个数*5

    4核  : 4个、5个进程 —— 20条线程/进程   : 80-100个任务

本文参与 腾讯云自媒体同步曝光计划,分享自作者个人站点/博客。
原始发表:2018/05/17 ,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 作者个人站点/博客 前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体同步曝光计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
相关产品与服务
云数据库 MySQL
腾讯云数据库 MySQL(TencentDB for MySQL)为用户提供安全可靠,性能卓越、易于维护的企业级云数据库服务。其具备6大企业级特性,包括企业级定制内核、企业级高可用、企业级高可靠、企业级安全、企业级扩展以及企业级智能运维。通过使用腾讯云数据库 MySQL,可实现分钟级别的数据库部署、弹性扩展以及全自动化的运维管理,不仅经济实惠,而且稳定可靠,易于运维。
领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档