Python学习总结笔记（3）--多线程与线程同步

多线程与线程同步网上讲的很多了，这里就简单总结下。

很多地方都讲了Python的多线程实际上是“假的”，原因就是Python的底层实现有一个GIL锁：Global Interpreter Lock，任何Python线程执行前，必须先获得GIL锁，然后，每执行100条字节码，解释器就自动释放GIL锁，让别的线程有机会执行。这个GIL全局锁实际上把所有线程的执行代码都给上了锁，所以，多线程在Python中只能交替执行，即使100个线程跑在100核CPU上，也只能用到1个核。

所以在Python中，可以使用多线程，但不要指望能有效利用多核。

0x01 线程处理扩展类

Python中使用线程有两个模块，低级的thread和封装thread的高级threading，建议用threading，还可以自己再扩展threading。简单模型如下：

#/usr/bin/env python
#coding:utf-8

__author__ = 'kikay'

import threading

class ThreadEx(threading.Thread):
def __init__(self):
super(ThreadEx,self).__init__()

def run(self):
pass

需要重载run方法即可。下面是一个具体的例子：

# /usr/bin/env python
# coding:utf-8

__author__ = 'kikay'

import threading
import random
import time

# 静态变量
Num = 0

class ThreadEx(threading.Thread):
def __init__(self,name):
super(ThreadEx, self).__init__()
self.__name=name

def run(self):
time.sleep(1)
global Num
print 'thread({0})is running...'.format(self.__name)
#运行10次
for i in range(10):
Num+=1

if __name__ == '__main__':
t = ThreadEx('t1')
t.start()
print 'main is running ...'
t.join()
print 'Num is ',Num

子线程对Num进行了10次自增操作，所以最终Num的值为10。

0x02 线程同步

现在多个子线程同时对Num进行自增操作，但是保证Num的值不大于10：

# /usr/bin/env python
# coding:utf-8

__author__ = 'kikay'

import threading
import random
import time

# 静态变量
Num = 0

class ThreadEx(threading.Thread):
def __init__(self,name):
super(ThreadEx, self).__init__()
self.__name=name

def run(self):
#time.sleep(1)
global Num
print 'thread({0})is running...'.format(self.__name)
#运行10次
for i in range(10):
if Num<10:
time.sleep(0.1)
Num=Num+1
print "thread({0})'s Num is {1}".format(self.__name,Num)

if __name__ == '__main__':
threadNum=10
threads=[]

for i in range(threadNum):
t = ThreadEx('t{0}'.format(i))
t.daemon=False
threads.append(t)

for i in range(threadNum):
threads[i].start()

print 'main is running ...'

for i in range(threadNum):
threads[i].join()

print 'Num is ',Num

运行结果如下：

thread(t0)is running...
thread(t1)is running...
thread(t2)is running...
thread(t3)is running...
thread(t4)is running...
thread(t5)is running...
thread(t6)is running...
thread(t7)is running...
thread(t8)is running...
thread(t9)is running...
main is running ...
thread(t0)'s Num is 1
thread(t2)'s Num is 3thread(t1)'s Num is 3

thread(t3)'s Num is 4thread(t4)'s Num is 5

thread(t5)'s Num is 6
thread(t6)'s Num is 7
thread(t7)'s Num is 8
thread(t8)'s Num is 9
thread(t9)'s Num is 10
thread(t0)'s Num is 11
thread(t1)'s Num is 12
thread(t2)'s Num is 13
thread(t4)'s Num is 14
thread(t5)'s Num is 15
thread(t3)'s Num is 16
thread(t6)'s Num is 17
thread(t7)'s Num is 18
thread(t8)'s Num is 19
Num is  19

结果显示，最终Num的值为19，大于了10。这是因为没有做线程同步造成的：

if Num<10:
time.sleep(0.1)
Num=Num+1
print "thread({0})'s Num is {1}".format(self.__name,Num)

上面的代码在某一时刻有一个子线程进行了Num<10判断，进入了if判断下的语句，但是没有进行自增运算，在这个时间空隙中，可能还有多个子线程也进入了if下的语句，从而导致最终Num的值大于了10。

Python中提供了“锁”的对象了实现简单的线程同步。建议大家使用RLock这个模块（原因见http://blog.csdn.net/cnmilan/article/details/8849895），修改过的代码如下：

# /usr/bin/env python
# coding:utf-8

__author__ = 'kikay'

import threading
import random
import time

# 静态变量
Num = 0
lock=threading.RLock()

class ThreadEx(threading.Thread):
def __init__(self,name):
super(ThreadEx, self).__init__()
self.__name=name

def run(self):
#time.sleep(1)
global Num
global lock
print 'thread({0})is running...'.format(self.__name)
#运行10次
for i in range(10):
lock.acquire()
try:
if Num<10:
time.sleep(0.1)
Num=Num+1
print "thread({0})'s Num is {1}".format(self.__name,Num)
finally:
lock.release()

if __name__ == '__main__':
threadNum=10
threads=[]

for i in range(threadNum):
t = ThreadEx('t{0}'.format(i))
t.daemon=False
threads.append(t)

for i in range(threadNum):
threads[i].start()

print 'main is running ...'

for i in range(threadNum):
threads[i].join()

print 'Num is ',Num

运行结果如下：

thread(t0)is running...
thread(t1)is running...
thread(t2)is running...
thread(t3)is running...
thread(t4)is running...
thread(t5)is running...
thread(t6)is running...
thread(t7)is running...
thread(t8)is running...
thread(t9)is running...
main is running ...
thread(t0)'s Num is 1
thread(t1)'s Num is 2
thread(t2)'s Num is 3
thread(t3)'s Num is 4
thread(t4)'s Num is 5
thread(t5)'s Num is 6
thread(t6)'s Num is 7
thread(t7)'s Num is 8
thread(t8)'s Num is 9
thread(t9)'s Num is 10
Num is  10

0x03 条件同步

上面的“锁”实现了基本的同步，如果现在有5个自增线程和5个自减线程同时运行（无限循环），保证Num的值不能小于0，不能大于10，继续用上面的RLock，修改后的代码如下：

# /usr/bin/env python
# coding:utf-8

__author__ = 'kikay'

import threading
import random
import time

# 静态变量
Num = 0
lock=threading.RLock()

#自增线程类
class ThreadAddEx(threading.Thread):
def __init__(self,name):
super(ThreadAddEx, self).__init__()
self.__name=name
self.__working=True

def run(self):
time.sleep(0.1)
global Num
global lock
#print 'threadAdd({0})is running...'.format(self.__name)
while self.__working:
lock.acquire()
try:
if Num<10:
time.sleep(0.1)
Num=Num+1
print "threadAdd({0})'s Num is {1}".format(self.__name,Num)
finally:
lock.release()
def stop(self):
self.__working=False

#自减线程类
class ThreadSubEx(threading.Thread):
def __init__(self,name):
super(ThreadSubEx, self).__init__()
self.__name=name
self.__working=True

def run(self):
time.sleep(0.1)
global Num
global lock
#print 'threadSub({0})is running...'.format(self.__name)
while self.__working:
lock.acquire()
try:
if Num>0:
time.sleep(0.1)
Num=Num-1
print "threadSub({0})'s Num is {1}".format(self.__name,Num)
finally:
lock.release()
def stop(self):
self.__working=False

if __name__ == '__main__':
threadNum=5
threads=[]

for i in range(threadNum):
t = ThreadAddEx('t{0}'.format(i))
t.daemon=False
threads.append(t)

for i in range(threadNum):
t = ThreadSubEx('t{0}'.format(i))
t.daemon=False
threads.append(t)

for i in range(len(threads)):
threads[i].start()

print 'main is running ...'

#运行1s
time.sleep(1)
#停止
for i in range(len(threads)):
threads[i].stop()

for i in range(len(threads)):
threads[i].join()

print 'Num is ',Num

运行结果：

main is running ...
threadAdd(t0)'s Num is 1
threadAdd(t3)'s Num is 2
threadSub(t0)'s Num is 1
threadAdd(t4)'s Num is 2
threadAdd(t2)'s Num is 3
threadSub(t1)'s Num is 2
threadAdd(t1)'s Num is 3
threadSub(t3)'s Num is 2
threadSub(t2)'s Num is 1
threadSub(t4)'s Num is 0
threadAdd(t0)'s Num is 1
threadAdd(t3)'s Num is 2
threadSub(t0)'s Num is 1
threadAdd(t4)'s Num is 2
threadAdd(t2)'s Num is 3
threadSub(t1)'s Num is 2
threadAdd(t1)'s Num is 3
threadSub(t3)'s Num is 2
Num is  2

实现了条件同步。但是上面的实现过程中，当不符合条件时，子线程会进入下一次循环，试想下，如果是10个自增线程，1个自减线程，那么自增线程将大量陷入“无效”循环中，不断lock.acquire和lock.release，我们能不能让子线程不符合条件时进行wait，然后当条件符合时，再唤醒该子线程继续执行呢？可以利用Python的Condition来实现：

# /usr/bin/env python
# coding:utf-8

__author__ = 'kikay'

import threading
import random
import time

# 静态变量
Num = 0
# lock=threading.RLock()
con = threading.Condition()

# 自增线程类
class ThreadAddEx(threading.Thread):
def __init__(self, name):
super(ThreadAddEx, self).__init__()
self.__name = name
self.__working = True

def run(self):
time.sleep(0.1)
global Num
# global lock
global con
# print 'threadAdd({0})is running...'.format(self.__name)
while self.__working:
con.acquire()
try:
time.sleep(0.1)
if Num < 5:
Num = Num + 1
print "threadAdd({0})'s Num is {1}".format(self.__name, Num)
# 肯定大于0，唤醒减线程
con.notify()
# 等待
else:
print 'threadAdd waiting ...'
con.wait()
finally:
con.release()

def stop(self):
self.__working = False

# 自减线程类
class ThreadSubEx(threading.Thread):
def __init__(self, name):
super(ThreadSubEx, self).__init__()
self.__name = name
self.__working = True

def run(self):
time.sleep(0.1)
global Num
# global lock
global con
# print 'threadSub({0})is running...'.format(self.__name)
while self.__working:
con.acquire()
try:
time.sleep(0.1)
if Num > 0:
Num = Num - 1
print "threadSub({0})'s Num is {1}".format(self.__name, Num)
# 肯定小于10了，唤醒自增线程
con.notify()
# 等待
else:
print 'threadSub waiting ...'
con.wait()
finally:
con.release()
def stop(self):
self.__working = False

if __name__ == '__main__':
threadNum = 10
threads = []

for i in range(threadNum):
t = ThreadAddEx('t{0}'.format(i))
t.daemon = False
threads.append(t)

for i in range(threadNum):
t = ThreadSubEx('t{0}'.format(i))
t.daemon = False
threads.append(t)

for i in range(len(threads)):
threads[i].start()

print 'main is running ...'

# 运行5s
time.sleep(5)
# 停止
for i in range(len(threads)):
threads[i].stop()

for i in range(len(threads)):
threads[i].join()

print 'Num is ', Num

运行结果：

main is running ...
threadAdd(t6)'s Num is 1
threadAdd(t4)'s Num is 2
threadAdd(t2)'s Num is 3
threadAdd(t0)'s Num is 4
threadAdd(t5)'s Num is 5
threadAdd waiting ...
threadAdd waiting ...
threadSub(t0)'s Num is 4
threadSub(t2)'s Num is 3
threadSub(t4)'s Num is 2
threadSub(t6)'s Num is 1
threadSub(t8)'s Num is 0
threadAdd(t9)'s Num is 1
threadSub(t1)'s Num is 0
threadSub waiting ...
threadSub waiting ...
threadSub waiting ...
threadSub waiting ...
threadAdd(t7)'s Num is 1
threadAdd(t8)'s Num is 2
threadAdd(t6)'s Num is 3
threadAdd(t4)'s Num is 4
threadAdd(t2)'s Num is 5
threadAdd waiting ...
threadAdd waiting ...
threadSub(t0)'s Num is 4
threadSub(t2)'s Num is 3
threadSub(t4)'s Num is 2
threadSub(t6)'s Num is 1
threadSub(t8)'s Num is 0
threadAdd(t9)'s Num is 1
threadSub(t1)'s Num is 0
threadAdd(t7)'s Num is 1
threadAdd(t8)'s Num is 2
threadAdd(t6)'s Num is 3
threadAdd(t4)'s Num is 4
threadAdd(t2)'s Num is 5
threadAdd waiting ...
threadSub(t0)'s Num is 4
threadSub(t2)'s Num is 3
threadAdd(t1)'s Num is 4
threadSub(t4)'s Num is 3
threadSub(t6)'s Num is 2
threadSub(t8)'s Num is 1
threadAdd(t9)'s Num is 2
threadSub(t1)'s Num is 1
threadSub(t3)'s Num is 0
threadSub waiting ...
threadAdd(t7)'s Num is 1
threadAdd(t8)'s Num is 2
threadSub(t7)'s Num is 1
threadAdd(t6)'s Num is 2
threadSub(t9)'s Num is 1
threadAdd(t4)'s Num is 2
threadAdd(t2)'s Num is 3
threadAdd(t0)'s Num is 4
threadSub(t0)'s Num is 3
threadSub(t2)'s Num is 2
threadAdd(t1)'s Num is 3
threadAdd(t5)'s Num is 4
threadSub(t4)'s Num is 3
threadSub(t6)'s Num is 2
threadSub(t8)'s Num is 1
threadAdd(t9)'s Num is 2
threadSub(t1)'s Num is 1
threadSub(t3)'s Num is 0
Num is  0

需要注意的是，Condition适合用于不同线程间的同步问题，即用notify“唤醒”的是其他线程，而不是本线程。

还有利用Queue模块实现FIFO同步以及其它一些同步的模块等等，网上例子很多了，这里就不再赘述了。