Python多线程
2017-11-23 本文已影响0人
cnkai
python中提供了函数和类两种方式使用多线程：
创建多线程
函数方式
import threading
import time
def run(sec):
print('%s 线程开始了！' %threading.current_thread().name)
time.sleep(sec)
print('%s 线程结束了！' %threading.current_thread().name)
if __name__ == '__main__':
print('主线程开始执行：', threading.current_thread().name)
s_time = time.time()
thread_list = []
for i in range(5):
t = threading.Thread(target=run, args=(i,))
thread_list.append(t)
for t in thread_list:
t.start()
for t in thread_list:
t.join()
print('主线程执行结束', threading.current_thread().name)
print('一共用时：', time.time()-s_time)
image
类方式
使用类方式需要写一个类，继承自threading.Thread类，然后重写run()方法。
import threading
import time
class MyThread(threading.Thread):
def __init__(self, sec):
super(MyThread,self).__init__()
self.sec = sec
def run(self):
print('%s 线程开始了！' %threading.current_thread().name)
time.sleep(self.sec)
print('%s 线程结束了！' %threading.current_thread().name)
if __name__ == '__main__':
print('主线程开始执行', threading.current_thread().name)
s_time = time.time()
my_thread_list = []
for i in range(5):
my_thread = MyThread(i)
my_thread_list.append(my_thread)
for i in my_thread_list:
i.start()
for i in my_thread_list:
i.join()
print('一共用时：', time.time()-s_time)
print('主线程结束执行', threading.current_thread().name)
image
线程锁
threading.Lock()
由于线程之间可以共享数据，而线程交替被送上CPU运行，这时很容易出现的一个问题就是，一个全局变量在被某一个线程修改时，可能还没有达成我们想要得到的结果，就被撤下CUP。这时，下一个被送上CUP的线程也需要取得这个变量的值，这时候，这个值的结果其实并不是我们期望的那个结果了。例子如下：
import threading
num = 0
def run(n):
global num
for i in range(100000):
num = num + n
num = num - n
if __name__ == '__main__':
t1 = threading.Thread(target=run, args=(50,))
t2 = threading.Thread(target=run ,args=(16,))
t1.start()
t2.start()
t1.join()
t2.join()
print(num)
image
我们期望的值是0，然而因为多线程的原因，num这个变量被加上n之后，还没有来得及减去n，立刻被撤下CUP，最终的结果并不是我们期望的那样。
因此，我们的目的是：在执行加和减的时候，不要被打断，python提供了线程锁来实现这样的目的，例子如下：
import threading
num = 0
lock = threading.Lock()
def run(n):
global num
for i in range(100000):
lock.acquire()
num = num + n
num = num - n
lock.release()
if __name__ == '__main__':
t1 = threading.Thread(target=run, args=(50,))
t2 = threading.Thread(target=run ,args=(16,))
t1.start()
t2.start()
t1.join()
t2.join()
print(num)
threading.Rlock()
RLock称之为可重入锁，它可以被同一个线程多次请求，使用了“拥有的线程”和“递归等级”的概念，处于锁定状态时，RLock被某个线程拥有。拥有RLock的线程可以再次调用acquire()，释放锁时需要调用release()相同次数。
与Lock的区别是：
- 从所属上，Lock属于全局，不被任何一个线程所拥有。Rlock一旦被某一个线程所获得，那么直到该锁被释放，都属于该线程。
- 从使用上，Lock只能被获得一次，然后释放，多次获得将会出现死锁。Rlock可以被多次获得，但是需要注意的是，获得多少次，就必须释放相对应的次数，Rlock有一个计数器，每获得一次，计数加一，每释放一次，计数减一，计数为0时，该锁属于释放状态。
条件（condition）
条件同步机制：顾名思义，一个线程等待某个特定条件，而另一个线程发出满足特定条件的信号。
下面使用条件同步机制来实现生产者消费者模型。
import threading
import random
import time
class Producer(threading.Thread):
def __init__(self, integers, condition):
super(Producer,self).__init__()
self.integers = integers
self.condition = condition
def run(self):
while True:
integer = random.randint(0, 1000)
self.condition.acquire() #获取条件锁
print('condition acquired by %s' % threading.current_thread().name)
self.integers.append(integer)
print('%d appended to list by %s' % (integer, threading.current_thread().name))
print('condition notified by %s' % threading.current_thread().name)
self.condition.notify() #唤醒消费者线程
print('condition released by %s' % self.name)
self.condition.release() #释放条件锁
time.sleep(1)
class Consumer(threading.Thread):
def __init__(self, integers, condition):
super(Consumer,self).__init__()
self.integers = integers
self.condition = condition
def run(self):
while True:
self.condition.acquire() #获取条件锁
print('condition acquired by %s' % self.name)
while True:
if self.integers:
integer = self.integers.pop()
print('%d popped from list by %s' % (integer, self.name))
break
print('condition wait by %s' % self.name)
self.condition.wait() #等待状态，等待被唤醒，才会继续执行
print('condition released by %s' % self.name)
self.condition.release() #最后释放条件锁
if __name__ == '__main__':
integers = []
condition = threading.Condition()
t1 = Producer(integers, condition)
t2 = Consumer(integers, condition)
t1.start()
t2.start()
t1.join()
t2.join()
image
先来看消费者模型，生产者生产的速率慢，消费者消费的速率快，消费者是一个死循环，一直从integers列表中取值，如果列表中一直有值的情况下，不需要过多解释，就是一个一直获得锁，弹值，释放锁的过程。关键的在于列表为空的情况下，使用了condition.wait()方法，此时，消费者处于休眠状态，相当于在这个地方停顿住，一旦出现condition.notify()，则该休眠状态结束，继续执行。
再来看生产者模型，生产者模型每生产一个元素，则调用condition.notify()方法来唤醒处于休眠状态的消费者，在释放锁，等待一秒钟，继续生产。
ThreadLocal
在多进程的模式下，每一个进程都有全局变量的一份副本，互相之间不会干扰。但是在多线程的模式下，全局变量对于每一个线程都是可见的，并且可以修改，有的时候我们希望线程使用自己的局部变量而不是使用全局变量，因为局部变量只有线程自己能看见，不会影响其他线程，而全局变量的修改必须加锁。
Python提供了threading.local()方法来实现上面的需求。
import threading
local_var = threading.local()
def get_name():
name = local_var.name
print('Hello, %s (in %s)' % (name, threading.current_thread().name))
def def_name(name):
local_var.name = name
get_name()
if __name__ == '__main__':
t1 = threading.Thread(target= def_name, args=('zhangsan',))
t2 = threading.Thread(target= def_name, args=('lisi',))
t1.start()
t2.start()
t1.join()
t2.join()
image
name变量此时相对于每一个线程都是独立的
但是在每个线程的内部都可以获得。
队列（Queue）
使用队列进行线程同步，不需要关心锁的问题，因为queue已经帮我们实现了
下面使用队列重写生产者消费者模型
import threading
import queue
import random
import time
class Producer(threading.Thread):
def __init__(self, queue):
super(Producer,self).__init__()
self.queue = queue
def run(self):
while True:
integer = random.randint(0, 1000)
self.queue.put(integer)
print('%d put to queue by %s' % (integer, self.name))
time.sleep(1)
class Consumer(threading.Thread):
def __init__(self, queue):
super(Consumer, self).__init__()
self.queue = queue
def run(self):
while True:
integer = self.queue.get()
print('%d popped from list by %s' % (integer, self.name))
self.queue.task_done()
if __name__ == '__main__':
queue = queue.Queue()
t1 = Producer(queue)
t2 = Consumer(queue)
t1.start()
t2.start()
t1.join()
t2.join()
