10-多线程、多进程和线程池编程
一知.半解 人气:0一、多线程、多进程和线程池编程
1.1、Python中的GIL锁
CPython中，global interpreter lock（简称GIL）是一个互斥体，用于保护对Python对象的访问，从而防止多个线程一次执行Python字节码（也就是说，GIL锁每次只能允许一个线程工作，无法多个线程同时在CPU上工作）。锁定是必要的，主要是因为CPython的内存管理不是线程安全的。（但是，由于存在GIL，因此其他功能已经变得越来越依赖于它所执行的保证。）CPython扩展必须支持GIL，以避免破坏线程。GIL之所以引起争议，是因为它在某些情况下阻止多线程CPython程序充分利用多处理器系统。请注意，潜在的阻塞或长时间运行的操作（例如I / O，图像处理和NumPy数字运算）发生在GIL 之外。因此，只有在GIL内部花费大量时间来解释CPython字节码的多线程程序中，GIL才成为瓶颈。但是即使不是瓶颈，GIL也会降低性能。总结这些：系统调用开销很大，尤其是在多核硬件上。两个线程调用一个函数所花的时间可能是单个线程两次调用该函数所花时间的两倍。GIL可以导致I / O绑定线程被调度在CPU绑定线程之前。并且它阻止了信号的传递。
import threading
num = 0
def add():
global num
for i in range(1000000):
num += 1
def desc():
global num
for i in range(1000000):
num -= 1
thread1 = threading.Thread(target=add)
thread2 = threading.Thread(target=desc)
thread1.start()#运行线程1
thread2.start()
thread1.join() #让线程1运行完毕才进行下一步
thread2.join()#让线程2运行完毕才进行下一步
print(num) #-197054 打印的结果应该是0，但是每次打印的结果都不一样，这说明线程没有按照要求一个一个运行完毕才进行下一个
#GIL会根据执行的字节码行数以及时间片释放GIL，GIL在遇到I/O的操作时候主动释放GIL
# （也就是说，当线程1遇到I/O操作的时候，会释放GIL切换到线程2运行）
1.2、多线程编程
线程：是指进程内的一个执行单元，同时也是操作系统即CPU执行任务的最小单位。因为Cpython的解释器上有一把全局锁即上面提到的GIL锁，一个进程中同一时间只允许一个线程执行，遇到I/O阻塞的时候，快速切换到另一个线程，线程也可以理解就是每当遇到I/O操作的时候，就会切换，节约的时间就是I/O操作的时间。
1.2.1：通过Thread类实例化：
threading.currentThread(): 返回当前的线程变量。 threading.enumerate(): 返回一个包含正在运行的线程的list。正在运行指线程启动后、结束前，不包括启动前和终止后的线程。 threading.activeCount(): 返回正在运行的线程数量，与len(threading.enumerate())有相同的结果。 Thread类提供了以下方法: run(): 用以表示线程活动的方法。 start():启动线程活动。 join([time]): 等待至线程中止。这阻塞调用线程直至线程的join() 方法被调用中止-正常退出或者抛出未处理的异常-或者是可选的超时发生。 isAlive(): 返回线程是否活动的。 getName(): 返回线程名。 setName(): 设置线程名。
Thread类的实例化：
import time
import threading
def get_detail_html(url):
print("get detail html started")
time.sleep(2)
print("get detail html end")
def get_detail_url(url):
print("get detail url started")
time.sleep(4)
print("get detail url end")
if __name__ == "__main__":
thread1 = threading.Thread(target=get_detail_html,args=("www.baidu.com",))
thread2 = threading.Thread(target=get_detail_url,args=("www.baidu.com",))
start_time = time.time()
thread1.start()
thread2.start()
print ("last time: {}".format(time.time()-start_time)) #0.00099945068359375
"""
get detail html started
get detail url startedlast time: 0.00099945068359375
get detail html end
get detail url end
"""
为什么上面的时间接近0秒，因为现在这个程序总共有三个线程，那三个线程呢？线程1、2以及主线程，按照main下面运行，就会发现线程1、2运行之后继续运行下面的print语句。
有没有就是当主线程运行完成之后就终止所有线程的呢？建立守护线程(setDaemon)这样主程序结束就会kill子线程。
if __name__ == "__main__":
thread1 = threading.Thread(target=get_detail_html,args=("www.baidu.com",))
thread2 = threading.Thread(target=get_detail_url,args=("www.baidu.com",))
thread1.setDaemon(True) # 是否守护线程
thread2.setDaemon(True)
start_time = time.time()
thread1.start()
thread2.start()
print ("last time: {}".format(time.time()-start_time))
"""
setDaemon()
参数一个布尔值，指示此线程是否是守护线程（真）（假）。必须在start()调用之前设置此参数，
否则RuntimeError引发该参数。它的初始值是从创建线程继承的；主线程不是守护程序线程，
因此在主线程中创建的所有线程默认为daemonic = False。 当没有活动的非守护线程时，整个Python程序将退出。 """
当所有线程运行结束后，主线程才结束：
if __name__ == "__main__":
thread1 = threading.Thread(target=get_detail_html,args=("www.baidu.com",))
thread2 = threading.Thread(target=get_detail_url,args=("www.baidu.com",))
start_time = time.time()
thread1.start()
thread2.start()
thread1.join()
thread2.join()
print ("last time: {}".format(time.time()-start_time))
1.2.2:通过继承Thread来实现多线程：（继承之后重写run方法，逻辑在run中进行）
import time
import threading
class GetDetailHtml(threading.Thread):
def __init__(self, name):
super().__init__(name=name)
def run(self):
print("get detail html started")
time.sleep(2)
print("get detail html end")
class GetDetailUrl(threading.Thread):
def __init__(self, name):
super().__init__(name=name)
def run(self):
print("get detail url started")
time.sleep(4)
print("get detail url end")
if __name__ == "__main__":
thread1 = GetDetailHtml("get_detail_html")
thread2 = GetDetailUrl("get_detail_url")
start_time = time.time()
thread1.start()
thread2.start()
thread1.join()
thread2.join()
print ("last time: {}".format(time.time()-start_time))
1.3、线程间通信-共享变量和Queue
线程间的通信方式第一种就是共享变量，共享变量就像上面第一个例子那样共享一个全局变量，但这种共享变量有缺点，这是线程不安全的状态。结果与预期值不一样。那么有没有一种线程安全的方式呢？当然有，那就是queue--同步队列类，Python 的 Queue 模块中提供了同步的、线程安全的队列类，包括FIFO（先入先出)队列Queue，LIFO（后入先出）队列LifoQueue，和优先级队列 PriorityQueue。该queue模块实现了多生产者，多消费者队列（生产者消费者模型）。当必须在多个线程之间安全地交换信息时，它在线程编程中特别有用。这些队列都实现了锁原语，能够在多线程中直接使用，可以使用队列来实现线程间的同步。
在这里就只设计到queue.Queue（maxsize = 0 ），maxsize是一个整数，用于设置可以放入队列中的项目数的上限。一旦达到此大小，插入将被阻塞，直到消耗队列项目为止。如果maxsize小于或等于零，则队列大小为无限。其他另外两个队列(后入先出)LifoQueue、(优先级队列)PriorityQueue根据需求来选择队列。
Queue.qsize() #返回队列的大小 Queue.empty() #如果队列为空，返回True,反之False Queue.full() #如果队列满了，返回True,反之False Queue.full #与 maxsize 大小对应 Queue.get([block[, timeout]]) #获取队列，timeout等待时间 Queue.get_nowait() #相当Queue.get(False) Queue.put(item) #写入队列，timeout等待时间 Queue.put_nowait(item) #相当Queue.put(item, False) Queue.task_done() #在完成一项工作之后，Queue.task_done()函数向任务已经完成的队列发送一个信号 Queue.join() #实际上意味着等到队列阻塞执行完毕为空，再执行别的操作
#通过queue的方式进行线程间同步
from queue import Queue
import time
import threading
def get_detail_html(queue):
while True:
#文章详情页
url = queue.get()
print("get detail html started")
time.sleep(2)
print("get url：{url}".format(url=url))
def get_detail_url(queue):
# 文章列表页
print("get detail url started")
time.sleep(1)
for i in range(20):
queue.put("http://cnblogs.com/lishuntao/{id}".format(id=i))
queue.task_done()#需要配合queue.join()使用
print("get detail url end")
if __name__ == "__main__":
queue = Queue(maxsize=1000)# maxsize是一个整数，用于设置可以放入队列中的项目数的上限。一旦达到此大小，插入将被阻塞，直到消耗队列项目为止。
thread_detail_url = threading.Thread(target=get_detail_url,args=(queue,))#将实例化的Queue作为参数传入
thread_detail_url.start()
for i in range(5): #五个线程共用数据
html_thread = threading.Thread(target=get_detail_html,args=(queue,))
html_thread.start()
start_time = time.time()
queue.join()#阻塞等待队列中任务全部处理完毕，需要配合queue.task_done使用
print("last time: {}".format(time.time()-start_time))
1.4、线程同步-Lock、RLock
什么叫线程同步？线程同步就是一个线程运行完成之后在进入下一个线程。上面第一个例子为啥num值不等于零？按理来说，先加100万次，再减去100万次，最后的结果是0。不等于0的原因是，将代码编译成字节码，前面load值以及运算值都没有出现问题，因为GIL锁会在I/O操作释放切换到其他线程，或者在特定的运行字节码行数的时候进行切换，然而上一个函数字节码的num值，很有可能两个线程会被共用值，赋值给desc函数的num值，因此会出此这样的情况，每次值都会不一样。那怎么解决这种情况呢？为了保证数据的正确性，需要对多个线程进行同步。使用 Thread 对象的 Lock 和 Rlock 可以实现简单的线程同步，这两个对象都有 acquire 方法和 release 方法，对于那些需要每次只允许一个线程操作的数据，可以将其操作放到 acquire 和 release 方法之间。但是用锁也会有缺点：1、用锁会影响性能。2、用锁会造成死锁（死锁循环以及两次acquire锁，没有释放锁）
from threading import Lock
import threading
total = 0
lock = Lock()
def add():
global lock
global total
for i in range(1000000):
lock.acquire()
total += 1
lock.release()
def desc():
global total
global lock
for i in range(1000000):
lock.acquire()
total -= 1
lock.release()
thread1 = threading.Thread(target=add)
thread2 = threading.Thread(target=desc)
thread1.start()
thread2.start()
thread1.join()
thread2.join()
print(total) #0
from threading import RLock
import threading
total = 0
lock = RLock() #在同一个线程里面，Rlock可以连续的调用acquire多次。一定要注意acquire的次数要和release的次数相等
def add():
global lock
global total
for i in range(1000000):
lock.acquire()
lock.acquire()
total += 1
lock.release()
lock.release()
def desc():
global total
global lock
for i in range(1000000):
lock.acquire()
total -= 1
lock.release()
thread1 = threading.Thread(target=add)
thread2 = threading.Thread(target=desc)
thread1.start()
thread2.start()
thread1.join()
thread2.join()
print(total)
#1. 用锁会影响性能
#2. 锁会引起死锁
#死锁的情况 1、A（a，b） 2、就是两次acquire ，第一次acquire没有将锁释放，第二次就不能获取到锁
"""
A(a、b) 当A先获取a，然后B获取到b，A就不能获取到b，B不能获取到A
acquire (a) 就这样进入到死循环即死锁的一种。
acquire (b)
B(a、b)
acquire (b)
acquire (a)
"""
1.5、线程同步-Condition使用
Condition（条件变量）里面有enter和exit两个魔法方法，因此可以利用with语句实现。with语句相当于就是一次获取锁以及释放锁的过程。条件变量总是与某种锁定相关联。可以传入，也可以默认创建一个。（当多个条件变量必须共享相同的锁时，传递一个输入很有用。）条件变量具有acquire()和release() 方法，它们调用关联锁的相应方法。它还具有一个wait()方法以及notify()和 notifyAll()方法。只有在调用线程获得了锁之后(with self.condition)，才必须调用这三个对象。
wait() #方法释放锁，然后块直到其被唤醒一个notify（）或notifyAll的（）调用在另一个线程相同的条件变量。唤醒后，它将重新获取锁并返回。也可以指定超时。 notify() #方法唤醒等待条件变量的线程中的一个，如果有的话正在等待。所述notifyAll的（） 方法唤醒等待条件变量的所有线程。
注意：notify()和notifyAll()方法不会释放锁；
一个线程将获得独占资源的锁去访问共享资源,通过生产者/消费者可以更好的描述这种方式，生产者添加一个随机数字到公共列表，而消费者将这个数字在公共列表中清除。看一下生产者类，生产者获得一个锁，添加一个数字，然后通知消费者线程有一些东西可以来清除，最后释放锁定。在各项动作不段切换的时候，将会触发不定期的随机暂停。
import threading
#通过condition完成协同读诗
class XiaoAi(threading.Thread):
def __init__(self, cond):
super().__init__(name="小爱")
self.cond = cond
def run(self):
self.cond.acquire()#与with self.cond:语句一样的效果__enter__,:(self.cond.acquire()) __exit__:（self.cond.release()）
self.cond.wait()
print("{} : 在 ".format(self.name))
self.cond.notify()
self.cond.wait()
print("{} : 好啊 ".format(self.name))
self.cond.notify()
self.cond.wait()
print("{} : 君住长江尾 ".format(self.name))
self.cond.notify()
self.cond.wait()
print("{} : 共饮长江水 ".format(self.name))
self.cond.notify()
self.cond.wait()
print("{} : 此恨何时已 ".format(self.name))
self.cond.notify()
self.cond.wait()
print("{} : 定不负相思意 ".format(self.name))
self.cond.notify()
self.cond.release()
class TianMao(threading.Thread):
def __init__(self, cond):
super().__init__(name="天猫精灵")
self.cond = cond
def run(self):
with self.cond:
print("{} : 小爱同学 ".format(self.name))
self.cond.notify()
self.cond.wait()
print("{} : 我们来对古诗吧 ".format(self.name))
self.cond.notify()
self.cond.wait()
print("{} : 我住长江头 ".format(self.name))
self.cond.notify()
self.cond.wait()
print("{} : 日日思君不见君 ".format(self.name))
self.cond.notify()
self.cond.wait()
print("{} : 此水几时休 ".format(self.name))
self.cond.notify()
self.cond.wait()
print("{} : 只愿君心似我心 ".format(self.name))
self.cond.notify()
self.cond.wait()
if __name__ == "__main__":
condition = threading.Condition()
xiaoai = XiaoAi(condition)
tianmao = TianMao(condition)
#在调用with condition之后才能调用wait或者notify方法
#condition有两层锁， 一把底层锁会在线程调用了wait方法的时候释放，
# 上面的锁会在每次调用wait的时候分配一把并放入到condition的等待队列中，等到notify方法的唤醒
#启动顺序很重要
xiaoai.start()
tianmao.start()
1.6、线程同步-Semaphore使用
信号量管理一个内部计数器，该计数器由每个acquire()调用递减， 并由每个release() 调用递增。计数器永远不能低于零。当acquire() 发现它为零时，它将阻塞，等待其他线程调用release()。即可以利用它来控制爬虫每次的请求次数，一次请求过多，会被禁止，为了反反爬就可以设置线程请求的并发数。
#Semaphore 是用于控制进入数量的锁
#文件， 读、写， 写一般只是用于一个线程写，读可以允许有多个
import threading
import time
class HtmlSpider(threading.Thread):
def __init__(self, url, sem):
super().__init__()
self.url = url
self.sem = sem
def run(self):
time.sleep(2)
print("got html text success")
self.sem.release()#释放一个信号量，使内部计数器增加一。当它在进入时为零
# 并且另一个线程正在等待它再次变得大于零时，唤醒该线程。
class UrlProducer(threading.Thread):
def __init__(self, sem):
super().__init__()
self.sem = sem
def run(self):
for i in range(20):
self.sem.acquire() #不带参数调用时：如果内部计数器在输入时大于零，则将其递减1并立即返回。
# 如果在进入时为零，则阻塞，等待其他线程调用 release使之大于零。（要互锁需要的代码块）
html_thread = HtmlSpider("https://baidu.com/{}".format(i), self.sem)
html_thread.start()
if __name__ == "__main__":
sem = threading.Semaphore(3) #设置每次运行数是3
url_producer = UrlProducer(sem)
url_producer.start()
1.7、ThreadPoolExecutor线程池
为什么要使用ThreadPoolExecutor？ ThreadPoolExecutor提供了一个简单的抽象，围绕多个线程并使用这些线程以并发方式执行任务。在正确的上下文中使用线程时，向您的应用程序添加线程可以帮助极大地提高应用程序的速度。通过使用多个线程，我们可以加快面对基于I/O操作型的应用程序，网络爬虫就是一个很好的例子。Web爬取网页程序通常会执行很多繁重的基于I / O的任务，例如获取和解析网站，如果我们要以同步方式获取每个页面，您会发现程序的主要瓶颈就是从互联网上获取这些页面 。通过使用诸如ThreadPoolExecutor之类的东西，我们可以通过同时执行多个读取并在返回每个页面时对其进行处理来有效地缓解此瓶颈。
创建ThreadPoolExecutor实例：
import time
from concurrent.futures import ThreadPoolExecutor
#未来对象，task的返回容器
#主线程中可以获取某一个线程的状态或者某一个任务的状态，以及返回值
#当一个线程完成的时候我们主线程能立即知道
#futures可以让多线程和多进程编码接口一致
def get_html(times):
time.sleep(times)
print("get page {} success".format(times))
return times
pool = ThreadPoolExecutor(max_workers=2) #max_workers 最大同时并发数，默认是操作系统的核的数量
#通过submit函数提交执行的函数到线程池中, submit 是立即返回
task1 = pool.submit(get_html, (3))
task2 = pool.submit(get_html, (2))
#要获取已经成功的task的返回
#done方法用于判定某个任务是否完成，完成返回True，没有完成返回False
print(task1.done()) #False
print(task2.cancel()) #False
time.sleep(4)
print(task1.done()) #True
#result方法可以获取task的执行结果（即函数的返回结果）
print(task1.result())
上下文管理器实例化：
import time
from concurrent.futures import ThreadPoolExecutor
def task(n):
time.sleep(3)
print("Processing {}".format(n))
def main():
print("Starting ThreadPoolExecutor")
#上下文管理器实例化ThreadPoolExecutor(线程池)对象
with ThreadPoolExecutor(max_workers=3) as executor:
for i in range(4):
future = executor.submit(task,(i))
print("All tasks complete")
if __name__ == '__main__':
main()
from concurrent.futures import ThreadPoolExecutor, wait, FIRST_COMPLETED
import time
def get_html(times):
time.sleep(times)
print("get page {} success".format(times))
return times
executor = ThreadPoolExecutor(max_workers=2)
#要获取已经成功的task的返回
urls = [3,2,4]
all_task = [executor.submit(get_html, (url)) for url in urls]
wait(all_task, return_when=FIRST_COMPLETED) #当第一个线程完成的时候，继续运行主线程
print("main")
更多方法调用详情在官网https://www.python.orghttps://img.qb5200.com/download-x/dev/peps/pep-3148/
1.8、多线程和多进程对比
多线程与多进程的对比，当遇到I/O操作的时候（例如爬虫，读文件等），多线程的速度优于多进程。当遇到计算密集型操作的时候（耗费CPU的操作，例如计算），多进程优于多线程。
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from concurrent.futures import ProcessPoolExecutor
#1、计算密集型（运用多进程）
def fib(n):
if n<=2:
return 1
return fib(n-1)+fib(n-2)
if __name__ == "__main__":
with ProcessPoolExecutor(3) as executor:
all_task = [executor.submit(fib, (num)) for num in range(25,40)]
start_time = time.time()
for future in as_completed(all_task):#返回一个迭代器，跟map()不同，这个迭代器的迭代顺序依照all_task返回（线程结束）的顺序。
data = future.result()
print("exe result: {}".format(data))
print("last time is: {}".format(time.time()-start_time))
#2. 对于io操作来说，多线程优于多进程
def random_sleep(n):
time.sleep(n)
return n
if __name__ == "__main__":
with ThreadPoolExecutor(3) as executor:
all_task = [executor.submit(random_sleep, (num)) for num in [2]*30]
start_time = time.time()
for future in as_completed(all_task):
data = future.result()
print("exe result: {}".format(data))
print("last time is: {}".format(time.time()-start_time))
"""
executor.map(func, list)
第一个参数为只接受一个参数的函数，后一个为可迭代对象。
这个map方法会把对函数的调用映射到到多个线程中。并返回一个future的迭代器。
"""
1.9、multiprocessing多进程编程
import multiprocessing
import time
def get_html(n):
time.sleep(n)
print("sub_progress success")
return n
if __name__ == "__main__":
progress = multiprocessing.Process(target=get_html, args=(2,))
print(progress.pid) #None
progress.start()
print(progress.pid) #10940
progress.join()
print("main progress end")
多进程编程：
import multiprocessing
#多进程编程
import time
def get_html(n):
time.sleep(n)
print("sub_progress success")
return n
if __name__ == "__main__":
#使用进程池
pool = multiprocessing.Pool(multiprocessing.cpu_count())
result = pool.apply_async(get_html, args=(3,)) #不用等待当前进程执行完毕，随时根据系统调度来进行进程切换。
#等待所有任务完成
pool.close() #告诉主进程，你等着所有子进程运行完毕后在运行剩余部分。
pool.join() #close必须在join前调用
print(result.get()) #3
imap/imap_unordered：
if __name__ == "__main__":
#使用线程池
pool = multiprocessing.Pool(multiprocessing.cpu_count())
for result in pool.imap(get_html, [1,5,3]): #按照映射的顺序输出
print("{} sleep success".format(result))
# for result in pool.imap_unordered(get_html, [1,5,3]): #谁先运行完成就运行谁
# print("{} sleep success".format(result))
2.0、进程间通信-Queue、Pipe，Manager（共享内存）
多进程之间的通信，不能运用多线程提供的queue.Queue。为了解决这个，多进程自己提供了一个multiprocessing.Queue。进程的用法和线程用法类似。多进程之间是不能共享全局变量的，然而多线程是可以共享全局变量的。多进程的数据是完全隔离的，当在linux/unix中fork数据的时候，在进程中的变量完全复制一份，复制到子进程中，这样两边的数据是互不影响的。还有multiprocessing中的queue不能用于pool进程池的。那么谁能用于进程池中呢？multiprocessing.Manager.Queue提供了可以用于进程池间的通信。
mutiprocessing.Queue（进程间的通信Queue）：
import time
from multiprocessing import Process, Queue
def producer(queue):
queue.put("a")
time.sleep(2)
def consumer(queue):
time.sleep(2)
data = queue.get()
print(data)
if __name__ == "__main__":
queue = Queue(10)
my_producer = Process(target=producer, args=(queue,))
my_consumer = Process(target=consumer, args=(queue,))
my_producer.start()
my_consumer.start()
my_producer.join()
my_consumer.join()
multiprocessing.Manager.Queue（进程池间通信Manager）：
import time
from multiprocessing import Pool, Manager
#multiprocessing中的queue不能用于pool进程池
#pool中的进程间通信需要使用manager中的queue
def producer(queue):
queue.put("a")
time.sleep(2)
def consumer(queue):
time.sleep(2)
data = queue.get()
print(data)
if __name__ == "__main__":
queue = Manager().Queue(10)
pool = Pool(2)
pool.apply_async(producer, args=(queue,))
pool.apply_async(consumer, args=(queue,))
#等待完成
pool.close()
pool.join()
Pipe（只能适用于两个进程之间的通信）：pipe性能高于queue
from multiprocessing import Process, Pipe
#通过pipe实现进程间通信
#pipe的性能高于queue
def producer(pipe):
pipe.send("lishuntao") #pipe发送数据
def consumer(pipe):
print(pipe.recv()) #pipe接收数据
if __name__ == "__main__":
recevie_pipe, send_pipe = Pipe() #实例化的时候是两个参数
#pipe只能适用于两个进程
my_producer = Process(target=producer, args=(send_pipe, ))
my_consumer = Process(target=consumer, args=(recevie_pipe,))
my_producer.start()
my_consumer.start()
my_producer.join()
my_consumer.join()
进程间通信其他方式（进程间共享内存）：
from multiprocessing import Process, Manager
def add_data(p_dict, key, value):
p_dict[key] = value
if __name__ == "__main__":
progress_dict = Manager().dict()
first_progress = Process(target=add_data, args=(progress_dict, "lishuntao", 22))
second_progress = Process(target=add_data, args=(progress_dict, "lishun", 18))
first_progress.start()
second_progress.start()
first_progress.join()
second_progress.join()
print(progress_dict) #{'lishun': 18, 'lishuntao': 22}
加载全部内容