Python第十二周学习总结

并行,并发,多线程

一、并发

 

1、基本概念

并发和并行的区别:

1)并行,parallel

同时做某些事,可以互不干扰的同一时刻做几件事。(解决并发的一种方法)

高速公路多个车道,车辆都在跑。同一时刻。

2)并发 concurrency

同时做某些事,一个时段内有事情要处理。(遇到的问题)

高并发,同一时刻内,有很多事情要处理。

2、并发的解决

1)队列、缓冲区

排队就是把人排成队列,先进先出,解决了资源使用的问题。

排成的队列,其实就是一个缓冲地带,就是缓冲区。

Queue模块的类queue、lifoqueue、priorityqueue。

2)争抢的

会有一个人占据窗口,其他人会继续争抢,可以锁定窗口,窗口不在为其他人服务,这就是锁机制。(锁的概念,排他性锁,非排他性锁)。

 

3)预处理

一种提前加载用户需要的数据的思路,预处理思想,缓存常用。

 

4)并行

日常可以通过购买更多的服务器,或者开多线程,实现并行处理,来解决并发问题。

水平扩展思想。

如果在但CPU上处理,就不是并行了。

但是多数服务都是多CPU的,服务的部署就是多机、分布式的,都是并行处理。

(串行比并行快)

5)提速

提高单个CPU性能,或单个服务器安装更多的CPU

这就是一种垂直扩展思想。

 

6)消息中间件

例如地跌站外的九曲回肠的走廊,缓冲人流。

常见的消息中间件有RabbitMQ,ActiveMQ(Apache)、RocketMQ(Apache)。

 

3、进程和线程

在实现了线程的操作系统中,线程是操作系统能够进行运算调度的最小单位。他包含在进程中,是进程中的实际运作单位。一个程序执行实例就是一个进程。

 

进程(process)是计算机中的程序关于某数据集合上的一次运行活动,是系统进行资源分配和调度的基本单位,是操作系统结构的基础。

 

进程和程序的关系

程序是源代码编译后的文件,而这些文件存放在磁盘上。当程序被操作系统加载到内存

中,就是进程,进程中存放着指令和数据(资源),也是线程的容器。

 

Linux进程有父进程、子进程,Windows的进程是平等关系。

 

线程,有时被称为轻量级进程,是程序执行流的最小单元,一个标准的线程由线程ID,当前指令指针(pc)寄存器集合堆栈组成。每个线程有自己独立的栈。

在许多系统中,创建一个线程比创建一个进程快10-100倍。

 

进程、线程的理解

现代操作系统提出的进程的概念,每一个进程都认为自己是独占所有的计算机硬件资源。

进程就是独立的王国,进程间不可以随便的共享数据。

线程就是省份,同一个进程内的线程可以共享进程的资源,每一个线程拥有自己独立的堆栈。

 

4、线程状态

状态 含义
就绪(ready) 线程能够运行,但在等待被调度,可能线程刚刚创建启动,或刚刚从阻塞恢复,或者被其他线程抢占。
运行(running) 线程正在运行
阻塞(Blocked) 线程等待外部事件发生而无法运行,如I/O操作。
终止(Terminated) 线程完成,或退出,或被取消。

图片19

5、Python中的线程和进程

进程会启动一个解释器进程,线程会共享一个解释器进程。

 

1)Python的线程开发

Python的线程开发使用标准库threading

 

2)Thread类

签名:

def __init__(self, group: None = …,
target: Optional[Callable[…, None]] = …,
name: Optional[str] = …,
args: Iterable = …,
kwargs: Mapping[str, Any] = …,
*, daemon: Optional[bool] = …) -> None: …

参数名 含义
target 线程调用对象,就是目标函数
name 为线程起名字
args 为目标函数传递实参,元组
Kwargs 为目标函数关键词传参,字典

 

3)线程启动

import threading
import time

def worker():
print(‘before’)
time.sleep(3)
print(‘finished’)

t = threading.Thread(target=worker)  #线程对象
t.start()   #启动

 

通过threading.Thread创建一个线程对象,target是目标函数,name可以指定名称。

需要调用start方法启动函数。

线程之所以执行函数,是因为线程中就是用来执行代码的,所以还是函数调用。

 

函数执行完毕后,线程也就退出了。

如果想让一个线程一直工作,不让线程退出就要利用到while循环。

import threading
import time

def worker():
count = 0
while True:
count += 1
print(‘before’)
time.sleep(3)
if count >5:
print(‘finished’)
break

t = threading.Thread(target=worker)  #线程对象
t.start()   #启动

 

4)线程退出

Python中没有提供终止线程的方法。线程在下面情况下退出。

  • 线程函数内语句执行完毕
  • 线程函数中抛出未处理的异常。

import threading
import time

def worker():
count = 0
while True:
if count >5:
break
#return
#raise RuntimeError(count)
time.sleep(3)
print(‘before’)
count += 1
print(‘finished’)

t = threading.Thread(target=worker)  #线程对象
t.start()   #启动
print(‘end’)

 

线程没有优先级,没有线程组的概念。也不能被销毁、停止、挂起,那么就是没有恢复和中断了。

5)线程的传参

import threading
import time

def add(x,y):
print(‘{}+{}={}’.format(x,y,x+y))

t1 = threading.Thread(target=add,name=’1′,args=(4,5))
t1.start()
time.sleep(2)

t2 = threading.Thread(target=add,name = ‘2’,args=(4,),kwargs={‘y’:6})
t2.start()
time.sleep(2)
t3 = threading.Thread(target=add,name=’3′,kwargs={‘x’:4,’y’:7})
t3.start()

 

线程中的传参,和函数传参没有什么区别,本质上就是函数传承。

 

6)threading的属性和方法

名称 含义
current_thread() 返回当前主线程
main_thread() 返回主线程对象
active_count() 当前处于alive状态的线程个数
enumerate() 返回所有活着的线程的列表,不包括已经终止的线程和未开始的线程
git_ident() 返回当前线程的ID,非0整数。

active_count、enumerate方法返回的值还包括主线程。

import threading
import time

def showinfo():
print(‘currentthread = {}’.format(threading.current_thread()))
print(‘main thread = {}’.format(threading.main_thread()))
print(‘active count = {}’.format(threading.active_count()))

def worker():
count = 0
showinfo()
while True:
if count>5:
break
time.sleep(5)
count += 1
print(‘finsh’)

t = threading.Thread(target=worker,name=’work’)
showinfo()
t.start()

print(‘end’)

 

 

currentthread = <_MainThread(MainThread, started 4048)>

main thread = <_MainThread(MainThread, started 4048)>

active count = 1

currentthread = <Thread(work, started 9084)>

end

main thread = <_MainThread(MainThread, stopped 4048)>

active count = 2

finsh

finsh

finsh

finsh

finsh

Finsh

 

名称 含义
Name 他只是一个名字,只是一个标识符,名字可以重名,getname()获取,setname()设置这个名词
Ident 线程id,是非0的整数,线程启动后才会有ID,否则为None,线程退出,此id依旧可以访问,此id可以重复访问。
Is_alive() 返回线程是否或者

线程的name只是一个名称,可以重复;id必须唯一,但可以在线程退出后在利用。

 

import threading

import time

 

 

def worker():

count = 0

while True:

if count > 5:

break

time.sleep(2)

count += 1

print(threading.current_thread().name)

 

t = threading.Thread(name=’work’,target=worker)

print(t.ident)

t.start()

 

while True:

time.sleep(1)

if t.is_alive():

print(‘{}{}alive’.format(t.name,t.ident))

else:

print(‘{}{}dead’.format(t.name,t.ident))

名称 含义
Start() 启动线程,每一个线程必须且只能执行该方法一次
Run() 运行线程函数

 

Start()启动线程,只能执行一次。操作系统。开辟新的线程。

Run()直接做的是主线程。函数调用。

(1)start()

import threading
import time

def worker():
count = 0
while True:
if count > 5:
break
time.sleep(3)
count += 1
print(‘running’)
class Mythread(threading.Thread):
def start(self):
print(‘start—-‘)
super().start()

def run(self):
print(‘run—-‘)
super().run()

t = Mythread(target=worker,name=’work’)

t.start()

start方法运行结果是start—-

run—-

Running

按照线程进行执行。

(2)run()

import threading
import time

def worker():
count = 0
while True:
if count>3:
break
time.sleep(2)
count += 1
print(‘runing’)
class Mythread(threading.Thread):
def start(self):
print(‘start—-‘)
super().start()

def run(self):
print(‘run—-‘)
super().run()

t = Mythread(target=worker,name=’work1’)
t.run()

# run—-
# runing

总结:run()执行结果就是直接是函数,调用,调用run函数。

Start()方法会调用run()方法,而run()方法可以运行函数。

 

  • start和run的区别

Start方法启动线程,启动了一个新的线程,名字叫做worker运行,但是run方法,并没有启动新的线程,只是在主线程内调用了一个普通的函数。

 

7)多线程

 

多线程,一个进程中如果有多个线程,就是多线程,是先一种并发。

import threading
import time

def worker():
count = 0
while True:
if count>3:
break
time.sleep(2)
count += 1
print(‘runing’)
print(threading.current_thread().name,threading.current_thread().ident)
class Mythread(threading.Thread):
def start(self):
print(‘start—-‘)
super().start()

def run(self):
print(‘run—-‘)
super().run()

t1 = Mythread(target=worker,name=’work1′)
t2 = Mythread(target=worker,name=’work2’)

# t1.run()
# t2.run()
####runing
# MainThread 1380
# runing
# MainThread 1380
# runing
# MainThread 1380
t1.start()
t2.start()
# start—-
# run—-
# start—-
# run—-
# runing
# work2 5048
# runing
# work1 9048

 

Start()方法work1和work2交替执行。启动线程后,进程内多个活动的线程并行工作,就是多线程。

Run()方法中没有开启新的线程,就是普通函数调用,所以执行完t1.run()

,然后执行t2.run(),run()方法就不是多线程。

 

一个进程中至少有一个线程,并作为程序的入口,这个线程就是主线程,一个线程必须有一个主线程。

其他线程成为工作线程。

 

8)线程安全

import threading

def worker():
for x in range(100):
print(‘{}is running’.format(threading.current_thread().name))

for x in range(1,4):
name = ‘worker{}’.format(x)
t = threading.Thread(name=name,target=worker)
t.start()

 

利用ipython执行的结果是不是一行行的打印,而是很多字符串打印在了一起。

这样说明了print函数被打断了,被线程切换打断了,print函数分为两步,第一步是打印字符串,第二部是换行,就在这个期间,发生了线程的切换,说明了print函数是线程不安全的。

 

线程安全:线程执行一段代码,不会产生不确定的结果,那么这段代码是线程安全的。

 

解决上面打印的问题:

(1)不让print打印换行

import threading

 

def worker():

for x in range(100):

print(‘{} is running.\n’.format(threading.current_thread().name),end=”)

 

for x in range(1,5):

name = ‘worker{}’.format(x)

t = threading.Thread(name=name,target=worker)

t.start()

利用字符串是不可变类型,可以作为一个整体不可分割输出,end=’’就不在print输出换行了。

 

  • 使用logging

标准库里面的logging模块,是日志处理模块,线程安全的,生产环境代码都使用logging。

import threading
import logging

def worker():
for x in range(100):
# print(‘{} is running.\n’.format(threading.current_thread().name),end=”)
logging.warning(‘{}is running’.format(threading.current_thread().name))
for x in range(1,5):
name = ‘worker{}’.format(x)
t = threading.Thread(name=name,target=worker)
t.start()

 

 

 

9)daemon线程和non-daemon线程

daemon不是Linux里面的守护进程。

 

进程靠线程执行代码,至少有一个主线程,其他线程是工作线程。

主线程是第一个启动的线程。

父线程:如果A中启动了一个线程B,那么A就是B的父线程。

子线程:B就是A的子线程。

 

源码Thread的__init__ 方法中。

If deamon is not None:

Self._daemonic = daemon

else:

Self._daemonic = current_thread().daemon

Self._ident = None

线程daemon属性,如果设定就是用户的设置,否则,就取当前线程的daemon的值。

主线程是non-daemon线程,即daemon = False。

import time
import threading

def foo():
time.sleep(5)
for i in range(20):
print(i)

t = threading.Thread(target=foo,daemon=False)
t.start()
print(‘end’)

 

daemon设置False值,主线程执行完毕后,等待工作线程。

import time
import threading

def foo():
time.sleep(5)
for i in range(20):
print(i)

t = threading.Thread(target=foo,daemon=True)
t.start()
print(‘end’)

Daemon值改为true,主线程执行完毕后直接退出。

名称 含义
Daemon 表示线程是否是daemon,这个值必须在start()之前设置,否则引发RuntimeError异常
IsDaemon() 是否是daemon线程
SetDaemon 设置daemon线程,必须在start方法之前设置。

总结:线程阿具有一个daemon属性,可以显示设置为True或者False,也可以不设置,则取默认值None。

 

如果不设置daemon,就取当前线程的daemon来设置他。

主线程是non-daemon线程,即daemon = False。

从主线程创建的所有线程的不设置daemon属性,则默认daemon = False,也就是non-daemon线程。

程序在没有活着的non-daemon线程运行时推出,也是就剩下的只是daemon线程,主线程才能推出。否则主线程只能等待。

 

构造线程的时候,可以设置daemon属性,这个属性必须在start方法前设置好。

daemon=True主线程不等。工作线程

daemon=False主线程等。只要有一个non-daemon就会等待。

控制一个属性的。

在start之前。

 

只是有一个non-daemon就会等待,没有的话直接不等,直接结束线程。

总结:

线程具有daemon属性,可以设置为True或者False。

(激活的non-daemon,主线程才会等待工作线程。)

import time
import threading

def bar():
time.sleep(10)
print(‘bar’)

def foo():
for i in range(20):
print(i)
t = threading.Thread(target=bar,daemon=False)
t.start()
t = threading.Thread(target=foo,daemon=True)
t.start()

print(‘end’)

 

这样不会执行bar的,因为主线程的daemon设置的值为True,改为False就好了。

活着让主线程sleep几秒。

import time
import threading

def bar():
time.sleep(10)
print(‘bar’)

def foo(n):
for i in range(n):
print(i)
t1 = threading.Thread(target=foo,args=(10,),daemon=True)
t1.start()
t2 = threading.Thread(target=foo,args=(20,),daemon=False)
t2.start()

time.sleep(6)
print(‘end’)

 

如果non-daemon线程的时候,主线程退出,也不会结束所有的daemon线程,直到所有的non-daemon线程全部结束,如果还有daemon线程,主线程需要退出,会结束所有的daemon线程,退出。

 

主线程是non-daemon。其他线程靠传参。

决定的是是否需要等待。如果有激活的non-daemon,就需要等待,没有激活的,主线程直接退出。

 

  • join方法

import time
import threading
def foo(n):
for i in range(n):
print(i)
time.sleep(1)

t1 = threading.Thread(target=foo,args=(10,),daemon=True)
t1.start()
t1.join()

 

 

利用join,主线程被迫等待他。把当前线程阻塞住了,x.join就等待谁。保证代码的执行顺序。

 

使用了join方法后,daemon线程执行完了,主线程才退出了。

 

Join(timeout= None),是线程的标准方法之一。

一个线程中调用另一个线程的join方法,调用者将被阻塞,直到被调用线程终止。

一个线程可以被join多次。

Timeout参数指定调用者等待多久,没有设置超时的,就会一直等待到调用的线程结束。

调用谁的join方法,就是join谁,就要等睡。

 

 

 

 

  • daemon线程应用场景

简单来说,本来并没有daemon Thread,这个概念唯一的作用是,当把一个线程设置为daemon,他会随着主线程的退出而退出。

主要应用场景为:

  • 后台任务。发送心跳包,监控。
  • 主线程工作才有用的线程。如主线程中维护着公共的资源,主线程已经清理了,准备退出,而工作线程使用这些资源工作没有意义了,一起退出最合适。
  • 随时可以被终止的线程。

 

如果主线程退出,想所有其他工作线程一起退出,就使用daemon=True来创建工作线程。

import time
import threading

def bar():
while True:
time.sleep(1)
print(‘bar’)

def foo():
print(‘t1 daemon = {}’.format(threading.current_thread().isDaemon()))
t2 = threading.Thread(target=bar)
t2.start()

print(‘t2 daemon = {}’.format(t2.isDaemon()))

t1 = threading.Thread(target=foo,daemon=True)
t1.start()

time.sleep(3)
print(‘Main end’)

 

改造成一直执行的:

import time
import threading

def bar():
while True:
time.sleep(1)
print(‘bar’)

def foo():
print(‘t1 daemon = {}’.format(threading.current_thread().isDaemon()))
t2 = threading.Thread(target=bar)
t2.start()
t2.join()
print(‘t2 daemon = {}’.format(t2.isDaemon()))

t1 = threading.Thread(target=foo,daemon=True)
t1.start()
t1.join()

time.sleep(3)
print(‘Main end’)

 

Daemon线程,简化了手动关闭线程的工作。

 

12)threading.local 类

 

局部变量的实现:

import threading
import time

def worker():
x = 0
for i in range(10):
time.sleep(0.01)
x += 1
print(threading.current_thread(),x)

for i in  range(10):
threading.Thread(target=worker).start()

 

利用全局变量实现:

import threading
import time

globals_data = threading.local()

def worker():
globals_data.x = 0
for i in range(10):
time.sleep(0.01)
globals_data.x += 1
print(threading.current_thread(),globals_data.x)

for i in  range(10):
threading.Thread(target=worker).start()

 

import threading

X = ‘abc’
ctx = threading.local()
ctx.x = 123

print(ctx,type(ctx),ctx.x)

def worker():
print(X)
print(ctx)
print(ctx.x)   #打印的时候出错,表示x不能跨线程
print(‘working’)

worker()
print()
threading.Thread(target=worker).start() #另一个线程启动

 

threading.local类构建了一个大字典,其元素是每一线程实例地址为key和线程对象引用线程单独的字典的映射。

 

通过threading.local实例就可在不同的线程中,安全的使用线程独有的数据,做到了线程间数据隔离,如同本地变量一样安全。

 

Local和线程相关的大字典,每次利用的时候利用线程的小字典来顶替local实例的大字典。

不利用的话,全局变量的话直接就是threading.local和本地线程相关的数据。

 

 

13)定时器timer延迟执行

Threading.Timer继承自thread,这个类用来另一多久执行一个函数。

Class threading.Timer(interval,function,args=None,kwargs=None)

Start方法执行以后,Timer对象会处于等待状态,等待了interval之后,开始执行function函数的。如果在执行函数之前的等待阶段,使用了cancel方法,就会跳过执行函数结果。

 

本质上就是一个Thread,只是没有提供name,daemon。

import threading
import logging
import time

def worker():
logging.info(‘in worker’)
time.sleep(2)

t = threading.Timer(5,worker)
t.start()  #启动
print(threading.enumerate())
t.cancel()   #取消
time.sleep(1)
print(threading.enumerate())

 

[<_MainThread(MainThread, started 7512)>, <Timer(Thread-1, started 6644)>]

[<_MainThread(MainThread, started 7512)>]

 

import threading
import logging
import time

def worker():
logging.info(‘in worker’)
time.sleep(2)

t = threading.Timer(5,worker)
t.cancel()   #取消
t.start()  #启动
print(threading.enumerate())
time.sleep(1)
print(threading.enumerate())

 

[<_MainThread(MainThread, started 7512)>]

[<_MainThread(MainThread, started 7512)>]

 

 

 

二、线程同步

1、概念

线程同步,线程间协同,通过某种技术,让一个线程访问某些数据时候,其他线程不能访问这些数据,直到该线程完成对数据的操作。

 

不同操作系统实现技术有所不同,有临界区、互斥量、信号量、事件Event。

 

2、Event

Event事件,是线程间通信机制中最简单的实现,使用一个内部的标记flag,通过flag的True或False的变化来进行操作。

名称 含义
set() 标记为True
clear() 标记为False
is_set() 标记是否为True
Wait(timeout=None) 设置等待标记为True的时长,None为无限等待,等到返回True,未等到超时了返回False。

 

课堂练习:老板雇佣了一个工人,让他生产杯子,老板一直等着这个工人,直到上产了十个杯子。

1)利用join

import threading
import time
import logging

def worker(count=10):
cups = []
while len(cups)<count:
logging.info(‘wprking’)
time.sleep(0.01)
cups.append(1)
print(len(cups))
logging.info(‘I am finished’)
w = threading.Thread(target=worker)
w.start()
w.join()

 

 

 

 

2)利用event

import threading
import logging
import time

def boss(event:threading.Event):
logging.info(‘I am boss,waiting’)
event.wait()
logging.info(‘good job’)

def worker(event:threading.Event,count=10):
logging.info(‘I am working for u’)
cups = []
while True:
logging.info(‘makeing’)
time.sleep(1)
cups.append(1)
if len(cups) >= count:
print(len(cups))
event.set()
break
logging.info(‘finished my job.cups={}’.format(cups))

event = threading.Event()
w = threading.Thread(target=worker,args=(event,))
b = threading.Thread(target=boss,args=(event,))
w.start()
b.start()

 

 

 

 

 

 

  • wait的应用

import threading
import logging
logging.basicConfig(level=logging.INFO)

def do(event:threading.Event,interval:int):
while not event.wait(interval):  #没有置set,所以是False。   不是False的时候就不能进入循环了。
logging.info(‘do sth’)    #没三秒打印一次。   not False执行此语句

e = threading.Event()
threading.Thread(target=do,args=(e,10)).start()

e.wait(12)  #整体停留了十秒。
e.set()    #重置为True。
print(‘end’)

 

  • 练习,实现timer。

 

总结:

使用同一个Event用来做标记。

Event的wait优于time.sleep,更快的切换到其他线程,提高并发效率。

 

 

import threading
import time

class MyTimer:
def __init__(self,interval,function,args,kwargs):
self.interval = interval
self.target = function
self.args = args
self.kwargs = kwargs
self.event = threading.Event()
self.thread = threading.Thread(target=self.target,args=self.args,kwargs=self.kwargs)

def start(self):
self.event.wait(self.interval)
if not self.event.is_set():   #如果没有置False,那么就是False,not False为True,执行run语句。
self.run()

def run(self):
self.start()

self.event.set()

def cancel(self):
self.event.set()

 

 

三、Lock锁

1)锁,凡是存在共享资源争抢的地方都可以使用锁。从而保证只有一个使用者可以完全使用这个资源。

 

lock.acquire  上锁    lock.release  解锁

import threading
import logging
import time
FORMAT = ‘%(asctime)s %(threadName)s %(thread)d %(message)s’
logging.basicConfig(format=FORMAT,level=logging.INFO)

cups = []
def worker(count=10):
logging.info(‘i am work’)
while len(cups) < count:
time.sleep(0.1)
cups.append(1)
logging.info(‘i am finsh.cups={}’.format(len(cups)))

for _ in range(10):
threading.Thread(target=worker,args=(1000,)).start()

 

2018-05-26 15:38:25,913 Thread-1 32 i am work

2018-05-26 15:38:25,913 Thread-2 4332 i am work

2018-05-26 15:38:25,913 Thread-3 9992 i am work

2018-05-26 15:38:25,914 Thread-4 8464 i am work

2018-05-26 15:38:25,914 Thread-5 9968 i am work

2018-05-26 15:38:25,915 Thread-6 8712 i am work

2018-05-26 15:38:25,915 Thread-7 4412 i am work

2018-05-26 15:38:25,915 Thread-8 8456 i am work

2018-05-26 15:38:25,915 Thread-9 8316 i am work

2018-05-26 15:38:25,915 Thread-10 9772 i am work

2018-05-26 15:38:35,925 Thread-8 8456 i am finsh.cups=1000

2018-05-26 15:38:36,023 Thread-7 4412 i am finsh.cups=1001

2018-05-26 15:38:36,023 Thread-1 32 i am finsh.cups=1002

2018-05-26 15:38:36,023 Thread-6 8712 i am finsh.cups=1003

2018-05-26 15:38:36,024 Thread-5 9968 i am finsh.cups=1004

2018-05-26 15:38:36,024 Thread-4 8464 i am finsh.cups=1005

2018-05-26 15:38:36,024 Thread-10 9772 i am finsh.cups=1006

2018-05-26 15:38:36,024 Thread-2 4332 i am finsh.cups=1007

2018-05-26 15:38:36,025 Thread-3 9992 i am finsh.cups=1008

2018-05-26 15:38:36,025 Thread-9 8316 i am finsh.cups=1009

 

运行结果来看,多线程调度,导致了判断失误,多生产了杯子只有用到了锁。

Lock,锁,一旦线程获得锁,其他要获得锁的线程将被阻塞。

名称 含义
acquire(blocking=True,timeout=-1) 默认阻塞,阻塞可以设置超时时间,非阻塞时,timeout禁止设置,成果获取锁,返回True,否则返回None
Release 释放锁,可以从任何线程调用释放,

已上锁的锁,会被重置到unlocked未上锁的锁上调用,抛出RuntimeError异常。

import threading
import logging
import time
FORMAT = ‘%(asctime)s %(threadName)s %(thread)d %(message)s’
logging.basicConfig(format=FORMAT,level=logging.INFO)

cups = []
lock = threading.Lock()
def worker(count=10):
logging.info(‘i am work’)
lock.acquire()
while len(cups) < count:
print(threading.current_thread(),len(cups))
time.sleep(0.000001)
cups.append(1)
logging.info(‘i am finsh.cups={}’.format(len(cups)))
lock.release()

for _ in range(10):
threading.Thread(target=worker,args=(1000,)).start()

 

上锁位置不对,由一个线程抢占,并独自占锁并完成任务。

import threading
import logging
import time
FORMAT = ‘%(asctime)s %(threadName)s %(thread)d %(message)s’
logging.basicConfig(format=FORMAT,level=logging.INFO)

cups = []
lock = threading.Lock()
def worker(count=10):
logging.info(‘i am work’)
flag= False
while True:
lock.acquire() #获取锁

if len(cups) >= count:
flag = True
# print(threading.current_thread(),len(cups))
time.sleep(0.000001)
if not flag:
cups.append(1)
print(threading.current_thread(),len(cups))
lock.release()   #追加后释放锁
if flag:
break
logging.info(‘i am finsh.cups={}’.format(len(cups)))

for _ in range(10):
threading.Thread(target=worker,args=(1000,)).start()

 

锁保证了数据完整性,但是性能下降好多。

If flag:break是为了保证release方法被执行,否则就出现了死锁,得到锁的永远没有释放。

 

计数器类,可以加可以减。

2)加锁、解锁

一般加锁就需要解锁,但是加锁后解锁前,还要有一些代码执行,就有可能抛出异常,一旦出现异常锁是无法释放的,但是当前线程可能因为这个就异常终止了,这就产生了死锁。

 

加锁。解锁常用语句:

  • 使用..finally语句保证锁的释放。
  • With上下文管理,锁对象支持上下文管理。

import threading
import time

class Counter:
def __init__(self):
self._val = 0
self.__lock = threading.Lock()

@property
def value(self):
return self._val

def inc(self):
try:
self.__lock.acquire()
self._val += 1
finally:
self.__lock.release()

def dec(self):
with self.__lock:
self._val -= 1

def run(c:Counter,count=1000):
for _ in range(10):
for i in range(-50,50):
if i<0:
c.dec()
else:
c.inc()

c = Counter()
c1 = 10
c2 = 10
for i in range(c1):
threading.Thread(target=run,args=(c,c2)).start()

while True:
time.sleep(1)
if threading.active_count() == 1:
print(threading.enumerate())
print(c.value)
break
else:
print(threading.enumerate())

 

 

不影响其他线程的切换,但是上锁后其他线程被阻塞了。只能等待。

 

 

 

3)锁的应用场景

适用于访问和修改同一个共享资源的时候,读写同一个资源的时候。

 

全部是读取同一个共享资源需要锁吗?

因为共享资源是不可变的,每一次读取都是一样的值,所以不用加锁。

 

使用锁的注意事项:

 

少用锁必要时用锁,使用了锁,多线程访问被锁的资源时候,就成了串行,要么排队执行,要么争抢执行。

 

加锁时间越短越好,不需要拍就立即释放锁。

一定要避免死锁。

 

不使用锁,有了效率,但是结果是错的。

使用了锁,效率低下,但是结果是对的。

 

4)非阻塞锁使用

import threading
import logging
import time

FORMAT = ‘%(asctime)s %(threadName)s %(thread)d %(message)s’
logging.basicConfig(format=FORMAT,level=logging.INFO)

def worker(tasks):
for task in tasks:
time.sleep(0.01)
if task.lock.acquire(False):
logging.info(‘{}{}begin to start’.format(threading.current_thread(),task.name))
else:
logging.info(‘{}{}is working’.format(threading.current_thread(),task.name))

class Task:
def __init__(self,name):
self.name = name
self.lock = threading.Lock()

tasks = [Task(‘task-{}’.format(x))for x in range(10)]

for i in range(5):
threading.Thread(target=worker,name=’worker-{}’.format(i),args=(tasks,)).start()

 

 

5)可重入锁RLock:

是线程相关的锁

线程A可重复锁,并可以多次成功获取,不会阻塞 ,最后要在线程A中做和acquire次数相同的release。

 

拿到这把锁的线程可以多次使用。

别的线程拿到的话也是被阻塞的。

一个线程占用锁的时候,其他线程不能拿到,只能的是阻塞。直到当前线程次有的锁全部释放完,其他线程才可以获取。

 

可重入锁,与线程相关,可在一个线程中获取锁,并可继续在同一线程中不阻塞获取锁,当锁未释放完,其他线程获取锁就会阻塞。直到当前持有锁的线程释放完了锁。

 

四、Condition

构造方法:condition(lock=None),可以传入一个lock对象或Rlock对象,默认是Rlock。

名称 含义
Acquire(*args) 获取锁
Wait(self,timeout=None) 等待超时
Notify(n=1) 唤醒之多指定书目个数的等待的线程,没有等待的线程就没有任何操作
Notify_all() 唤醒所有等待的线程。

 

用于生产者、消费者模型,为了解决生产者消费者速度匹配的问题:

 

import threading
import logging
import random

FORMAT = ‘%(asctime)s %(threadName)s %(thread)d %(message)s’
logging.basicConfig(format=FORMAT,level=logging.INFO)

class Dispatcher:
def __init__(self):
self.data = None
self.event = threading.Event()

def produce(self,total):
for _ in range(total):
data = random.randint(0,100)
logging.info(data)
self.data = data
self.event.wait(1)
self.event.set()

def consume(self):
while not self.event.is_set():
data = self.data
logging.info(‘recieved{}’.format(data))
self.data = None
self.event.wait(0.5)

d = Dispatcher()
p = threading.Thread(target=d.produce,args=(10,),name=’producer’)
c = threading.Thread(target=d.consume,name=’consume’)
c.start()
p.start()

 

消费者采用主动消费,消费者浪费了大量的时间,主动来查看有没有数据。换成通知的机制。

import threading
import logging
import random

FORMAT = ‘%(asctime)s %(threadName)s %(thread)d %(message)s’
logging.basicConfig(format=FORMAT,level=logging.INFO)

class Dispatcher:
def __init__(self):
self.data = None
self.event = threading.Event()
self.cond = threading.Condition()

def produce(self,total):
for _ in range(total):
data = random.randint(0,100)
with self.cond:
logging.info(data)
self.data = data
self.cond.notify_all()
self.event.wait(1)
self.event.set()

def consume(self):
while not self.event.is_set():
with self.cond:
self.cond.wait()
logging.info(‘recieved{}’.format(self.data))
self.data = None
self.event.wait(0.5)

d = Dispatcher()
p = threading.Thread(target=d.produce,args=(10,),name=’producer’)
c = threading.Thread(target=d.consume,name=’consume’)
c.start()
p.start()

 

如果是一个生产者,多个消费者呢:

import threading
import logging
import random

FORMAT = ‘%(asctime)s %(threadName)s %(thread)d %(message)s’
logging.basicConfig(format=FORMAT,level=logging.INFO)

class Dispatcher:
def __init__(self):
self.data = None
self.event = threading.Event()
self.cond = threading.Condition()

def produce(self,total):
for _ in range(total):
data = random.randint(0,100)
with self.cond:
logging.info(data)
self.data = data
self.cond.notify_all()
self.event.wait(1)  #模拟生产速度
self.event.set()

def consume(self):
while not self.event.is_set():
with self.cond:
self.cond.wait()  #阻塞等通知
logging.info(‘recieved{}’.format(self.data))
self.data = None
self.event.wait(0.5)  #模拟消费 的速度

d = Dispatcher()
p = threading.Thread(target=d.produce,args=(10,),name=’producer’)

for i in range(5):
c = threading.Thread(target=d.consume, name=’consume{}’.format(i))
c.start()
p.start()

 

 

Self.cond.notify_all()发通知:

修改为self.cond.notify(n=2)  随机通知两个消费者。

Condition总结:

用于生产者消费者模型中,解决生产者,消费者速度匹配的问题。

采用了通知机制,非常有效率。

 

使用方式:

使用condition,必须先acquire,用完了要release。因为内部实现了锁,默认使用了RLock锁。最好的方式就是使用上下文。

消费者wait,等待通知。

生产者生产好消息,对消费者发出通知,可以使用notify或者notify_all方法。

 

本文来自投稿,不代表Linux运维部落立场,如若转载,请注明出处:http://www.178linux.com/99570

发表评论

登录后才能评论

This site uses Akismet to reduce spam. Learn how your comment data is processed.

联系我们

400-080-6560

在线咨询:点击这里给我发消息

邮件:1823388528@qq.com

工作时间:周一至周五,9:30-18:30,节假日同时也值班