【python高级用法】线程

前言

Python通过标准库的 threading 模块来管理线程。这个模块提供了很多不错的特性，让线程变得无比简单。实际上，线程模块提供了几种同时运行的机制，实现起来非常简单。

线程模块

• 线程对象
• Lock对象
• RLock对象
• 信号对象
• 条件对象
• 事件对象

 简单调用方式

class threading.Thread(group=None,target=None,name=None,args=(),kwargs={})

• group: 一般设置为 None ，这是为以后的一些特性预留的
• target: 当线程启动的时候要执行的函数
• name: 线程的名字，默认会分配一个唯一名字 Thread-N
• args: 传递给 target 的参数，要使用tuple类型
• kwargs: 同上，使用字典类型dict

 创建线程的方法非常实用，通过`target`参数、`arg`和`kwarg`告诉线程应该做什么。下面这个例子传递一个数字给线程（这个数字正好等于线程号码），目标函数会打印出这个数字。

import threadingdef function(i):print ("function called by thread %i\n" % i)returnthreads = []for i in range(5):t = threading.Thread(target=function , args=(i, ))threads.append(t)t.start()
for i in range(5):t.join()

 线程被创建之后并不会马上运行，需要手动调用 start() ， join() 让调用它的线程一直等待直到执行结束（即阻塞调用它的主线程， t 线程执行结束，主线程才会继续执行）

 实现一个线程

使用threading模块实现一个新的线程，需要下面3步：

• 定义一个 Thread 类的子类
• 重写 __init__(self [,args]) 方法，可以添加额外的参数
• 最后，需要重写 run(self, [,args]) 方法来实现线程要做的事情

 当你创建了新的 Thread 子类的时候，你可以实例化这个类，调用 start() 方法来启动它。线程启动之后将会执行 run() 方法。

为了在子类中实现线程，我们定义了 myThread 类。其中有两个方法需要手动实现：

import threading
import timeexitFlag = 0class myThread (threading.Thread):def __init__(self, threadID, name, counter):threading.Thread.__init__(self)self.threadID = threadIDself.name = nameself.counter = counterdef run(self):print("Starting " + self.name)print_time(self.name, self.counter, 5)print("Exiting " + self.name)def print_time(threadName, delay, counter):while counter:if exitFlag:# 译者注：原书中使用的thread，但是Python3中已经不能使用thread，以_thread取代，因此应该# import _thread# _thread.exit()thread.exit()time.sleep(delay)print("%s: %s" % (threadName, time.ctime(time.time())))counter -= 1# Create new threads
thread1 = myThread(1, "Thread-1", 1)
thread2 = myThread(2, "Thread-2", 2)# Start new Threads
thread1.start()
thread2.start()# 以下两行为译者添加，如果要获得和图片相同的结果，
# 下面两行是必须的。疑似原作者的疏漏
thread1.join()
thread2.join()
print("Exiting Main Thread")

threading 模块是创建和管理线程的首选形式。每一个线程都通过一个继承 Thread 类，重写 run() 方法来实现逻辑，这个方法是线程的入口。在主程序中，我们创建了多个 myThread 的类型实例，然后执行 start() 方法启动它们。调用 Thread.__init__ 构造器方法是必须的，通过它我们可以给线程定义一些名字或分组之类的属性。调用 start() 之后线程变为活跃状态，并且持续直到 run() 结束，或者中间出现异常。所有的线程都执行完成之后，程序结束。

join() 命令控制主线程的终止。

Lock用法

当两个或以上对共享内存的操作发生在并发线程中，并且至少有一个可以改变数据，又没有同步机制的条件下，就会产生竞争条件，可能会导致执行无效代码、bug、或异常行为。

竞争条件最简单的解决方法是使用锁。锁的操作非常简单，当一个线程需要访问部分共享内存时，它必须先获得锁才能访问。此线程对这部分共享资源使用完成之后，该线程必须释放锁，然后其他线程就可以拿到这个锁并访问这部分资源了。

很显然，避免竞争条件出现是非常重要的，所以我们要保证，在同一时刻只有一个线程允许访问共享内存。

然而，在实际使用的过程中，我们发现这个方法经常会导致一种糟糕的死锁现象。当不同的线程要求得到一个锁时，死锁就会发生，这时程序不可能继续执行，因为它们互相拿着对方需要的锁。

 为了简化问题，我们设有两个并发的线程（ 线程A 和 线程B )，需要 资源1 和 资源2 .假设 线程A 需要 资源1 ， 线程B 需要 资源2 .在这种情况下，两个线程都使用各自的锁，目前为止没有冲突。现在假设，在双方释放锁之前， 线程A 需要 资源2 的锁， 线程B 需要 资源1 的锁，没有资源线程不会继续执行。鉴于目前两个资源的锁都是被占用的，而且在对方的锁释放之前都处于等待且不释放锁的状态。

# -*- coding: utf-8 -*-import threadingshared_resource_with_lock = 0
shared_resource_with_no_lock = 0
COUNT = 100000
shared_resource_lock = threading.Lock()# 有锁的情况
def increment_with_lock():global shared_resource_with_lockfor i in range(COUNT):shared_resource_lock.acquire()shared_resource_with_lock += 1shared_resource_lock.release()def decrement_with_lock():global shared_resource_with_lockfor i in range(COUNT):shared_resource_lock.acquire()shared_resource_with_lock -= 1shared_resource_lock.release()# 没有锁的情况
def increment_without_lock():global shared_resource_with_no_lockfor i in range(COUNT):shared_resource_with_no_lock += 1def decrement_without_lock():global shared_resource_with_no_lockfor i in range(COUNT):shared_resource_with_no_lock -= 1if __name__ == "__main__":t1 = threading.Thread(target=increment_with_lock)t2 = threading.Thread(target=decrement_with_lock)t3 = threading.Thread(target=increment_without_lock)t4 = threading.Thread(target=decrement_without_lock)t1.start()t2.start()t3.start()t4.start()t1.join()t2.join()t3.join()t4.join()print ("the value of shared variable with lock management is %s" % shared_resource_with_lock)print ("the value of shared variable with race condition is %s" % shared_resource_with_no_lock)

 RLock用法

这种锁对比Lock有是三个特点：

1. 谁拿到谁释放。如果线程A拿到锁，线程B无法释放这个锁，只有A可以释放；

2. 同一线程可以多次拿到该锁，即可以acquire多次；

3. acquire多少次就必须release多少次，只有最后一次release才能改变RLock的状态为unlocked）

import threading
import timeclass Box(object):lock = threading.RLock()def __init__(self):self.total_items = 0def execute(self, n):Box.lock.acquire()self.total_items += nBox.lock.release()def add(self):Box.lock.acquire()self.execute(1)Box.lock.release()def remove(self):Box.lock.acquire()self.execute(-1)Box.lock.release()## These two functions run n in separate
## threads and call the Box's methods
def adder(box, items):while items > 0:print("adding 1 item in the box")box.add()time.sleep(1)items -= 1def remover(box, items):while items > 0:print("removing 1 item in the box")box.remove()time.sleep(1)items -= 1## the main program build some
## threads and make sure it works
if __name__ == "__main__":items = 5print("putting %s items in the box " % items)box = Box()t1 = threading.Thread(target=adder, args=(box, items))t2 = threading.Thread(target=remover, args=(box, items))t1.start()t2.start()t1.join()t2.join()print("%s items still remain in the box " % box.total_items)

 信号量用法

 信号量是由操作系统管理的一种抽象数据类型，用于在多线程中同步对共享资源的使用。本质上说，信号量是一个内部数据，用于标明当前的共享资源可以有多少并发读取。

在threading模块中，信号量的操作有两个函数，即 acquire() 和 release() 

• 每当线程想要读取关联了信号量的共享资源时，必须调用 acquire() ，此操作减少信号量的内部变量, 如果此变量的值非负，那么分配该资源的权限。如果是负值，那么线程被挂起，直到有其他的线程释放资源。
• 当线程不再需要该共享资源，必须通过 release() 释放。这样，信号量的内部变量增加，在信号量等待队列中排在最前面的线程会拿到共享资源的权限。

 存在的问题：

假设有两个并发的线程，都在等待一个信号量，目前信号量的内部值为1。假设第线程A将信号量的值从1减到0，这时候控制权切换到了线程B，线程B将信号量的值从0减到-1，并且在这里被挂起等待，这时控制权回到线程A，信号量已经成为了负值，于是第一个线程也在等待。

# -*- coding: utf-8 -*-"""Using a Semaphore to synchronize threads"""
import threading
import time
import random# The optional argument gives the initial value for the internal
# counter;
# it defaults to 1.
# If the value given is less than 0, ValueError is raised.
semaphore = threading.Semaphore(0)def consumer():print("consumer is waiting.")# Acquire a semaphoresemaphore.acquire()# The consumer have access to the shared resourceprint("Consumer notify : consumed item number %s " % item)def producer():global itemtime.sleep(10)# create a random itemitem = random.randint(0, 1000)print("producer notify : produced item number %s" % item)# Release a semaphore, incrementing the internal counter by one.# When it is zero on entry and another thread is waiting for it# to become larger than zero again, wake up that thread.semaphore.release()if __name__ == '__main__':for i in range (0,5) :t1 = threading.Thread(target=producer)t2 = threading.Thread(target=consumer)t1.start()t2.start()t1.join()t2.join()print("program terminated")

 queue线程通信

Python的threading模块提供了很多同步原语，包括信号量，条件变量，事件和锁。如果可以使用这些原语的话，应该优先考虑使用这些，而不是使用queue（队列）模块。队列操作起来更容易，也使多线程编程更安全，因为队列可以将资源的使用通过单线程进行完全控制，并且允许使用更加整洁和可读性更高的设计模式。 

 

Queue常用的方法有以下四个：

• put(): 往queue中放一个item
• get(): 从queue删除一个item，并返回删除的这个item
• task_done(): 每次item被处理的时候需要调用这个方法
• join(): 所有item都被处理之前一直阻塞

 

from threading import Thread, Event
from queue import Queue
import time
import random
class producer(Thread):def __init__(self, queue):Thread.__init__(self)self.queue = queuedef run(self) :for i in range(10):item = random.randint(0, 256)self.queue.put(item)print('Producer notify: item N° %d appended to queue by %s' % (item, self.name))time.sleep(1)class consumer(Thread):def __init__(self, queue):Thread.__init__(self)self.queue = queuedef run(self):while True:item = self.queue.get()print('Consumer notify : %d popped from queue by %s' % (item, self.name))self.queue.task_done()if __name__ == '__main__':queue = Queue()t1 = producer(queue)t2 = consumer(queue)t3 = consumer(queue)t4 = consumer(queue)t1.start()t2.start()t3.start()t4.start()t1.join()t2.join()t3.join()t4.join()

 

生产者使用 Queue.put(item [,block[, timeout]]) 来往queue中插入数据。Queue是同步的，在插入数据之前内部有一个内置的锁机制。

可能发生两种情况：

• 如果 block 为 True ， timeout 为 None （这也是默认的选项，本例中使用默认选项），那么可能会阻塞掉，直到出现可用的位置。如果 timeout 是正整数，那么阻塞直到这个时间，就会抛出一个异常。
• 如果 block 为 False ，如果队列有闲置那么会立即插入，否则就立即抛出异常（ timeout 将会被忽略）。本例中， put() 检查队列是否已满，然后调用 wait() 开始等待。

消费者从队列中取出整数然后用 task_done() 方法将其标为任务已处理。

消费者使用 Queue.get([block[, timeout]]) 从队列中取回数据，queue内部也会经过锁的处理。如果队列为空，消费者阻塞。

 

参考链接：

Python并行编程 中文版 — python-parallel-programming-cookbook-cn 1.0 文档