清爽网站模板,网站上传 文件夹结构,和平区网站制作,交三百能在网站上找兼职做的前言
Python通过标准库的 threading 模块来管理线程。这个模块提供了很多不错的特性#xff0c;让线程变得无比简单。实际上#xff0c;线程模块提供了几种同时运行的机制#xff0c;实现起来非常简单。
线程模块
线程对象Lock对象RLock对象信号对象条件对象事件对象 简单…前言
Python通过标准库的 threading 模块来管理线程。这个模块提供了很多不错的特性让线程变得无比简单。实际上线程模块提供了几种同时运行的机制实现起来非常简单。
线程模块
线程对象Lock对象RLock对象信号对象条件对象事件对象 简单调用方式
class threading.Thread(groupNone,targetNone,nameNone,args(),kwargs{})
group: 一般设置为 None 这是为以后的一些特性预留的target: 当线程启动的时候要执行的函数name: 线程的名字默认会分配一个唯一名字 Thread-Nargs: 传递给 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(targetfunction , 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(targetincrement_with_lock)t2 threading.Thread(targetdecrement_with_lock)t3 threading.Thread(targetincrement_without_lock)t4 threading.Thread(targetdecrement_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 Boxs 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(targetadder, args(box, items))t2 threading.Thread(targetremover, 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(targetproducer)t2 threading.Thread(targetconsumer)t1.start()t2.start()t1.join()t2.join()print(program terminated) queue线程通信
Python的threading模块提供了很多同步原语包括信号量条件变量事件和锁。如果可以使用这些原语的话应该优先考虑使用这些而不是使用queue队列模块。队列操作起来更容易也使多线程编程更安全因为队列可以将资源的使用通过单线程进行完全控制并且允许使用更加整洁和可读性更高的设计模式。 Queue常用的方法有以下四个
put(): 往queue中放一个itemget(): 从queue删除一个item并返回删除的这个itemtask_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 文档
