1. 线程锁
如果不控制多个线程对同一资源进行访问的话，会对数据造成破坏，使得线程运行的结果不可预期。因此要引进线程锁。
线程同步能够保证多个线程安全访问竞争资源，最简单的同步机制是引入互斥锁。 互斥锁为资源引入一个状态：锁定/非锁定。某个线程要更改共享数据时，先将其锁定，此时资源的状态为“锁定”，其他线程不能更改；直到该线程释放资源，将 资源的状态变成“非锁定”，其他的线程才能再次锁定该资源。互斥锁保证了每次只有一个线程进行写入操作，从而保证了多线程情况下数据的正确性。
未引入锁前：
import threading
import time
NUM = 0
class MyThread(threading.Thread):
def run(self):
global NUM
NUM += 1
time.sleep(0.5)
msg = self.name+‘ set num to ‘+str(NUM)
print(msg)
if __name__ == ‘__main__‘:
for i in range(5):
t = MyThread()
t.start()
out:
Thread-5 set num to 2
Thread-3 set num to 3
Thread-2 set num to 5
Thread-1 set num to 5
Thread-4 set num to 4
引入锁之后：
import threading
import time
NUM = 0
class MyThread(threading.Thread):
def run(self, l):
l.acquire()
global NUM
NUM += 1
time.sleep(0.5)
msg = self.name+‘ set num to ‘+str(NUM)
l.release()
print(msg)
if __name__ == ‘__main__‘:
lock = threading.Lock()
for i in range(5):
t = MyThread()
t.run(lock)
out:
Thread-1 set num to 1
Thread-2 set num to 2
Thread-3 set num to 3
Thread-4 set num to 4
Thread-5 set num to 5
加线程锁
可 以看出，加了线程锁之后，尝试任何次数的执行，最终结果都是一样的，消除了线程不安全的隐患。过程是这样的：我们先建立了一个 threading.Lock类对象lock,在run方法里，我们使用lock.acquire()获得了这个锁。此时，其他的线程就无法再获得该锁 了，他们就会阻塞在“l.acquire()”这里，直到锁被另一个线程释放l.release()。
上边的锁机制，叫做“互斥锁”。同一时间仅允许一个线程进行更改数据。而信号量(Semaphore)是同时允许一定数量的线程更改数据 ，比如厕所有3个坑，那最多只允许3个人上厕所，后面的人只能等里面有人出来了才能再进去。
信号量：
import threading,time
def run(n):
semaphore.acquire()
time.sleep(1)
print("run the thread: %s" %n)
semaphore.release()
if __name__ == ‘__main__‘:
num= 0
semaphore = threading.BoundedSemaphore(5) #最多允许5个线程同时运行
for i in range(20):
t = threading.Thread(target=run,args=(i,))
t.start()
如果多个线程要调用多个现象，而A线程调用A锁占用了A对象，B线程调用了B锁占用了B对象,A线程不能调用B对象，B线程不能调用A对象，于是一直等待。这就造成了线程“死锁”。
Threading模块中，也有一个类，RLock，称之为可重入锁。该锁对象内部维护着一个Lock和一个counter对象。counter对象记
录了acquire的次数，使得资源可以被多次require。最后，当所有RLock被release后，其他线程才能获取资源。在同一个线程
中，RLock.acquire可以被多次调用，利用该特性，可以解决部分死锁问题。
解决死锁问题：
class _RLock:
"""This class implements reentrant lock objects.
A reentrant lock must be released by the thread that acquired it. Once a
thread has acquired a reentrant lock, the same thread may acquire it
again without blocking; the thread must release it once for each time it
has acquired it.
"""
def __init__(self):
self._block = _allocate_lock()
self._owner = None
# 初始化一个计数器
self._count = 0
def __repr__(self):
owner = self._owner
try:
owner = _active[owner].name
except KeyError:
pass
return "<%s %s.%s object owner=%r count=%d at %s>" % (
"locked" if self._block.locked() else "unlocked",
self.__class__.__module__,
self.__class__.__qualname__,
owner,
self._count,
hex(id(self))
)
def acquire(self, blocking=True, timeout=-1):
"""Acquire a lock, blocking or non-blocking.
When invoked without arguments: if this thread already owns the lock,
increment the recursion level by one, and return immediately. Otherwise,
if another thread owns the lock, block until the lock is unlocked. Once
the lock is unlocked (not owned by any thread), then grab ownership, set
the recursion level to one, and return. If more than one thread is
blocked waiting until the lock is unlocked, only one at a time will be
able to grab ownership of the lock. There is no return value in this
case.
When invoked with the blocking argument set to true, do the same thing
as when called without arguments, and return true.
When invoked with the blocking argument set to false, do not block. If a
call without an argument would block, return false immediately;
otherwise, do the same thing as when called without arguments, and
return true.
When invoked with the floating-point timeout argument set to a positive
value, block for at most the number of seconds specified by timeout
and as long as the lock cannot be acquired. Return true if the lock has
been acquired, false if the timeout has elapsed.
"""
me = get_ident()
if self._owner == me:
# 每次调用acquire，计数器加1
self._count += 1
return 1
rc = self._block.acquire(blocking, timeout)
if rc:
self._owner = me
self._count = 1
return rc
__enter__ = acquire
def release(self):
"""Release a lock, decrementing the recursion level.
If after the decrement it is zero, reset the lock to unlocked (not owned
by any thread), and if any other threads are blocked waiting for the
lock to become unlocked, allow exactly one of them to proceed. If after
the decrement the recursion level is still nonzero, the lock remains
locked and owned by the calling thread.
Only call this method when the calling thread owns the lock. A
RuntimeError is raised if this method is called when the lock is
unlocked.
There is no return value.
"""
if self._owner != get_ident():
raise RuntimeError("cannot release un-acquired lock")
# 每次调用release,计数器减1
self._count = count = self._count - 1
if not count:
self._owner = None
self._block.release()
def __exit__(self, t, v, tb):
self.release()
# Internal methods used by condition variables
def _acquire_restore(self, state):
self._block.acquire()
self._count, self._owner = state
def _release_save(self):
if self._count == 0:
raise RuntimeError("cannot release un-acquired lock")
count = self._count
self._count = 0
owner = self._owner
self._owner = None
self._block.release()
return (count, owner)
def _is_owned(self):
return self._owner == get_ident()
python3中Rlock的实现代码
死锁发生的条件：
- 互斥条件：线程对资源的访问是排他性的，如果一个线程对占用了某资源，那么其他线程必须处于等待状态，直到资源被释放。
- 请求和保持条件：线程T1至少已经保持了一个资源R1占用,但又提出对另一个资源R2请求，而此时，资源R2被其他线程T2占用，于是该线程T1也必须等待，但又对自己保持的资源R1不释放。
- 不剥夺条件：线程已获得的资源，在未使用完之前，不能被其他线程剥夺，只能在使用完以后由自己释放。
- 环 路等待条件：在死锁发生时，必然存在一个“进程-资源环形链”，即：{p0,p1,p2,...pn},进程p0（或线程）等待p1占用的资 源，p1等待p2占用的资源，pn等待p0占用的资源。（最直观的理解是，p0等待p1占用的资源，而p1而在等待p0占用的资源，于是两个进程就相互等 待）
活锁：
活锁：是指线程1可以使用资源，但它很礼貌，让其他线程先使用资源，线程2也可以使用资源，但它很绅士，也让其他线程先使用资源。这样你让我，我让你，最后两个线程都无法使用资源。
2. 队列
Python中的队列，常用的包括以下四种：1.先进先出队列 2.后进先出队列. 3.优先级队列 4.双向队列。这四种队列的使用，在queue模块中已经定义好了，具体使用如下。
1） 先进先出
import queue # 定义队列最大长度 q = queue.Queue(maxsize=5) # q.empty() 判断队列是否为空 print(‘if q is empty: ‘, q.empty()) # 向队列中push元素 q.put(1) q.put(2) q.put(3) q.put(4) # 判断队列长度是否达到上限 print(‘if q is full:‘, q.full()) print(‘queue size :‘ , q.qsize()) q.put(5) print(‘if q is full:‘, q.full()) # 向队列中put数据，如果队列满，则立即抛出异常，不阻塞 q.put_nowait(6) # 向队列中put数据,默认为block为True,表示阻塞，timeout表示最大等待时间，如果这段时间内put失败，则抛出异常 q.put(7, block=True, timeout=2) print(‘if q is empty: ‘, q.empty()) # 返回当前队列元素个数 print(‘queue size :‘ , q.qsize()) # 从队列中获取单个数据 print(q.get()) print(q.get()) print(q.get()) print(q.get()) print(q.get()) # q.get(block, timeout) 默认block为True,表示是否阻塞。timeout表示阻塞最长时间，如果这段时间内无法获取元素，则抛出异常 print(q.get(block=True, timeout=2)) # block=False时，如果get异常，立即抛出错误 print(q.get(block=False)) # 从队列中获取元素，如果获取异常则抛出异常，不阻塞。 # print(q.get_nowait()) print(‘if q is empty: ‘, q.empty()) queue.Queue()用法
2） 后进先出
import queue q = queue.LifoQueue() q.put(1) q.put(2) q.put(3) print(‘first get: ‘, q.get()) print(‘second get: ‘, q.get()) print(‘third get: ‘, q.get()) out: first get: 3 second get: 2 third get: 1 queue.LifoQueue后进先出队列
3） 优先对级
import queue q1 = queue.PriorityQueue() # 如果优先级相同，谁先放进去，先取出谁 q1.put((1, ‘alex1‘)) q1.put((2, ‘alex2‘)) q1.put((1, ‘alex3‘)) print(q1.get()) print(q1.get()) print(q1.get()) out: (1, ‘alex1‘) (1, ‘alex3‘) (2, ‘alex2‘) queue.PriorityQueue()优先级队列
4） 双向队列
import queue q2 = queue.deque() q2.append(1) q2.append(2) # 从右侧增加数据 q2.append(3) q2.appendleft(4) # 从左侧增加数据 q2.extend([5,6]) # 从右侧增加一个可迭代对象数据 q2.extendleft([7]) # 从左侧增加一个可迭代对象数据 print(q2) q2.pop() # 从右侧取数据 q2.popleft() # 从左侧取数据 print(q2) out: deque([7, 4, 1, 2, 3, 5, 6]) deque([4, 1, 2, 3, 5]) queue.deque()双向队列
9. 生产者消费者模型
有一个或多个生产者生产某种类型的数据，并放置在缓冲区（可以是数 组也可以是队列等数据结构）中；有一个消费者可以从缓冲区中取数据，每次取一项；系统保 证避免对缓冲区的重复操作，也就是说在任何时候只有一个主体（生产者或消费者）可以访问缓冲区。问题要确保缓冲区不溢出，即当缓冲区满时，生成者不会继续 向其中添加数据；当缓冲区空时，消费者不会从中移走数据。该模型可以解决：阻塞问题和程序解耦。下面是一个吃包子和做包子的案例代码：
import queue
import threading
import time
q = queue.Queue(20)
def productor(arg):
while True:
q.put(str(arg) + ‘号厨师弄出来的包子‘)
def consumer(arg):
while True:
print( arg, q.get())
time.sleep(2)
for i in range(3):
t = threading.Thread(target=productor, args=(i,))
t.start()
for j in range(20):
t = threading.Thread(target=consumer, args=(j,))
t.start()
生产者、消费者模型案例代码
生产者的工作是产生一块数据，放到buffer中，如此循环。与此同时，消费者在消耗这些数据（例如从buffer中把它们移除），每次一块。这里的关键词是“同时”。所以生产者和消费者是并发运行的，我们需要对生产者和消费者做线程分离。假设这样一种情况，生产者负责往队列中存数据，消费者负责从队列中取数据。这两个线程同时运行，会存在这样一种情况：在某一时间点，消费者把所有东西消耗完毕而生产者还在挂起（sleep）。消费者尝试继续进行消耗，但此时队列为空，出现异常。我们把这个实现作为错误行为（wrong behavior）。测试代码如下：
from threading import Thread, Lock
import time
import random
queue = []
lock = Lock()
class ProducerThread(Thread):
def run(self):
nums = range(5) #Will create the list [0, 1, 2, 3, 4]
global queue
while True:
num = random.choice(nums) #Selects a random number from list [0, 1, 2, 3, 4]
lock.acquire()
queue.append(num)
print("Produced", num)
lock.release()
time.sleep(random.random())
class ConsumerThread(Thread):
def run(self):
global queue
while True:
lock.acquire()
if not queue:
print("Nothing in queue, but consumer will try to consume")
exit()
num = queue.pop(0)
print("Consumed", num)
lock.release()
time.sleep(random.random())
ProducerThread().start()
ConsumerThread().start()
out:
Produced 1
Consumed 1
Produced 0
Produced 4
Produced 1
Consumed 0
Consumed 4
Produced 2
Consumed 1
Consumed 2
Produced 3
Consumed 3
Nothing in queue, but consumer will try to consume
错误行为代码
那么，正确行为应该是怎样的呢？
当队列中没有任何数据的时候，消费者应该停止运行并等待 (wait），而不是继续尝试进行消耗。而当生产者在队列中加入数据之后，应该有一个渠 道去告诉（notify）消费者。然后消费者可以再次从队列中进行消耗，而IndexError不再出现。而条件（Condition）正是用来解决这一 问题。
3. 进程
python
中的多线程其实并不是真正的多线程，如果想要充分地使用多核CPU的资源，在python中大部分情况需要使用多进程。Python提供了非
常好用的多进程包multiprocessing，只需要定义一个函数，Python会完成其他所有事情。借助这个包，可以轻松完成从单进程到并发执行的
转换。multiprocessing支持子进程、通信和共享数据、执行不同形式的同步，提供了Process、Queue、Pipe、Lock等组件。
1） 基本使用
from multiprocessing import Process
def foo(arg):
print(‘say hi‘,arg)
if __name__ == "__main__":
for i in range(10):
p = Process(target=foo,args=(i,))
#p.daemon = True # 等同于线程的threading.Thread.setDaemon
p.start()
#p.join()
out:
say hi 0
say hi 2
say hi 4
say hi 1
say hi 3
say hi 6
say hi 8
say hi 7
say hi 5
say hi 9
多进程的基本使用
2） 进程锁
当多个进程需要访问共享资源的时候，Lock可以用来避免访问的冲突。
同 样的，进程锁也包含了其他类型，包括RLock,Semaphore（用来控制对共享资源的访问数量，例如池的最大连接数）。Event。同 线程也是一样的。所属模块不同而 已：multiprocessing.Lock()，multiprocessing.RLock()，multiprocessing.Semaphore(n) ，multiprocessing.Event()。 具体使用案例可以查看http://www.cnblogs.com/kaituorensheng/p/4445418.html 或者将上边讲到的线程案例进行修改。
3） 进程间资源共享
默认情况下，进程间的数据是不可以进行共享的。但是可以通过以下三个方法进行数据共享：queues, Array, Manager.dict()
- queues
from multiprocessing import Process
from multiprocessing import queues
import multiprocessing
def foo(i,arg):
arg.put(i)
print(‘say hi‘,i,arg.qsize())
if __name__ == "__main__":
li = queues.Queue(20,ctx=multiprocessing) # 源码中执行了ctx.lock
for i in range(10):
p = Process(target=foo,args=(i,li,))
p.start()
out:
say hi 0 1
say hi 2 2
say hi 1 4
say hi 3 4
say hi 6 7
say hi 5 7
say hi 4 7
say hi 7 8
say hi 9 9
say hi 8 10
multiprocessing.queues
- Array（不常用）
数组：Array的特点：1.它的内存地址是连续的，而列表不是 2. 数组中元素的数据类型，在创建的时候就已经定义好了。3. 个数一定。在创建的时候就需要指定
from multiprocessing import Process
from multiprocessing import Array
def foo(i,arg):
arg[i] = i + 100
for item in arg:
print(item)
print(‘================‘)
if __name__ == "__main__":
li = Array(‘i‘, 5) # 创建一个最大包含5个元素的数组
for i in range(5):
p = Process(target=foo,args=(i,li,))
p.start()
out:
0
0
================
101
0
================
101
0
================
101
103
================
101
103
================
multiprocessing.Array
- Manager.dict（常用）
from multiprocessing import Process
from multiprocessing import Manager
def foo(i,arg):
arg[i] = i + 100
print(arg.values())
if __name__ == "__main__":
obj = Manager()
li = obj.dict()
for i in range(5):
p = Process(target=foo,args=(i,li,))
#p.daemon = True
p.start()
p.join()
out:
[100]
[100, 101]
[100, 101, 102]
[100, 101, 102, 103]
[100, 101, 102, 103, 104]
Manager.dict实现数据共享
4. 进程池
进程池在multiprocessing模块中已经定义好，只需要直接使用即可。主要包含2个方法apply和apply_async
- 进程池的使用：
from multiprocessing import Pool
import time
def f1(arg):
print(arg,‘b‘)
time.sleep(5)
print(arg,‘a‘)
if __name__ == "__main__":
pool = Pool(5)
# 定义30个任务
for i in range(30):
# pool.apply(func=f1,args=(i,))
pool.apply_async(func=f1,args=(i,))
# pool.close() # 等待所有的任务执行完毕后，修改进程状态为close。否则会阻塞到这里
time.sleep(2)
pool.terminate() # 立即终止全部子进程
pool.join() # 主进程在这里等待子进程全部执行完毕
out:
b
b
b
b
b # 执行到一半已经被强制终止
Process finished with exit code 0
进程池的使用
apply(self, func, args=(), kwds={}) # 使用arg和kwds参数调用func函数，结果返回前会一直阻塞，这样就导致子进程会顺序执行，而不是并发执行
apply_async(self, func, args=(), kwds={}, callback=None,error_callback=None)# apply()方法的一个变体，会返回一个结果对象。如果callback被指定，那么callback可以接收一个参数然后被调用，当结果准备好回调 时会调用callback，调用失败时，则用error_callback替换callback。 Callbacks应被立即完成，否则处理结果的线程会被阻塞。对比apply方法，该方法不会阻塞，类似线程的setDaemon
pool.close() # 阻止更多的任务提交到pool，待任务完成后，工作进程会退出
pool.terminate() # 不管任务是否完成，立即停止工作进程。在对pool对象进程垃圾回收的时候，会立即调用terminate()。
pool.join() # wait工作线程的退出，在调用join()前，必须调用close() or terminate()。这样是因为被终止的进程需要被父进程调用wait（join等价与wait），否则进程会成为僵尸进程。
参考:https://docs.python.org/3/library/multiprocessing.html#module-multiprocessing.pool
5. 协程
1） 概念
线程和进程的操作是由程序触发系统接口，最后的执行者是系统；协程的操作则是程序员。
协程存在的意义：对于多线程应用，CPU通过切片的方式来切换线程间的执行，线程切换时需要耗时（保存状态，下次继续）。协程，则只使用一个线程，在一个线程中规定某个代码块执行顺序。
协程的适用场景：当程序中存在大量不需要CPU的操作时（IO），适用于协程；
2） 实现
协程的实现主要借助2个模块，greenlet和gevent(内部调用greenlet)
- gevent
import gevent
def fun1():
print("www.baidu.com") # 第一步
gevent.sleep(0)
print("end the baidu.com") # 第三步
def fun2():
print("www.zhihu.com") # 第二步
gevent.sleep(0)
print("end th zhihu.com") # 第四步
gevent.joinall([
gevent.spawn(fun1),
gevent.spawn(fun2),
])
gevent
- greenlet
import greenlet
def fun1():
print("12") # 第一步
gr2.switch()
print("56") # 第三步
gr2.switch()
def fun2():
print("34") # 第二步
gr1.switch()
print("78") # 第四步
gr1 = greenlet.greenlet(fun1)
gr2 = greenlet.greenlet(fun2)
gr1.switch()
greenlet