threading模块使用

  1. #对于io操作来说，多线程和多进程性能差别不大
  2. #1.通过Thread类实例化
  4. import time
  5. import threading
  7. def get_detail_html(url):
  8.     print("get detail html started")
  9.     time.sleep(2)
  10.     print("get detail html end")
  13. def get_detail_url(url):
  14.     print("get detail url started")
  15.     time.sleep(4)
  16.     print("get detail url end")
  19. #2. 通过集成Thread来实现多线程
  21. class GetDetailHtml(threading.Thread):
  22.     def __init__(self, name):
  23.         super().__init__(name=name)
  25.     def run(self):
  26.         print("get detail html started")
  27.         time.sleep(2)
  28.         print("get detail html end")
  31. class GetDetailUrl(threading.Thread):
  32.     def __init__(self, name):
  33.         super().__init__(name=name)
  35.     def run(self):
  36.         print("get detail url started")
  37.         time.sleep(4)
  38.         print("get detail url end")
  40. if  __name__ == "__main__":
  41.     thread1 = GetDetailHtml("get_detail_html")
  42.     thread2 = GetDetailUrl("get_detail_url")
  43.     start_time = time.time()
  44.     thread1.start()
  45.     thread2.start()
  47.     thread1.join()
  48.     thread2.join()
  50.     #当主线程退出的时候， 子线程kill掉
  51.     print ("last time: {}".format(time.time()-start_time))

theading 控制线程数量

  1. #Semaphore 是用于控制进入数量的锁
  2. #文件， 读、写， 写一般只是用于一个线程写，读可以允许有多个
  4. #做爬虫
  5. import threading
  6. import time
  8. class HtmlSpider(threading.Thread):
  9.     def __init__(self, url, sem):
  10.         super().__init__()
  11.         self.url = url
  12.         self.sem = sem
  14.     def run(self):
  15.         time.sleep(2)
  16.         print("got html text success")
  17.         # 释放锁
  18.         self.sem.release()
  20. class UrlProducer(threading.Thread):
  21.     def __init__(self, sem):
  22.         super().__init__()
  23.         self.sem = sem
  25.     def run(self):
  26.         for i in range(20):
  27.             # 获取锁
  28.             self.sem.acquire()
  29.             html_thread = HtmlSpider("https://baidu.com/{}".format(i), self.sem)   # HtmlSpider调用
  30.             html_thread.start()
  32. if __name__ == "__main__":
  33.     sem = threading.Semaphore(3)   # 在20个任务中,保证一次只有3个线程同时工作
  34.     url_producer = UrlProducer(sem)
  35.     url_producer.start()

theading 线程同步

同步= 同步执行

  1. from threading import Lock, RLock, Condition #可重入的锁
  3. #在同一个线程里面，可以连续调用多次acquire， 一定要注意acquire的次数要和release的次数相等
  4. total = 0
  5. lock = RLock()   #RLock允许acquire重复不报错
  6. def add():
  7.     #1. dosomething1
  8.     #2. io操作
  9.     # 1. dosomething3
  10.     global lock
  11.     global total
  12.     for i in range(1000000):
  13.         lock.acquire()
  14.         lock.acquire()
  15.         total += 1
  16.         lock.release()
  17.         lock.release()
  20. def desc():
  21.     global total
  22.     global lock
  23.     for i in range(1000000):
  24.         lock.acquire()
  25.         total -= 1
  26.         lock.release()
  28. import threading
  29. thread1 = threading.Thread(target=add)
  30. thread2 = threading.Thread(target=desc)
  31. thread1.start()
  32. thread2.start()
  35. #
  36. thread1.join()
  37. thread2.join()
  38. print(total)
  40. #1. 用锁会影响性能
  41. #2. 锁会引起死锁
  42. #死锁的情况 A（a，b）
  43. """
  44. A(a、b)
  45. acquire (a)
  46. acquire (b)
  48. B(a、b)
  49. acquire (a)
  50. acquire (b)
  51. """

threading 线程间通信 -共享变量加锁

线程间通信用于:一个任务依赖另一个任务,生产者消费者关系
使用共享变量不加锁的话会出问题:
比如修改一个共享变量的值a,当2个线程拿到变量和修改变量是2步操作,第一步同时拿到a,a的值是一样的,导致修改后的a值不正确,a=1

  1. 1+1=2
  2. 1-1 =0
  3. 结果可能是2或者0
  4. 实际应该是1+1-1=1 ```python

    线程间通信

import time import threading from chapter11 import variables

from threading import Condition

1. 生产者当生产10个url以后就就等待，保证detail_url_list中最多只有十个url

2. 当url_list为空的时候，消费者就暂停

def get_detail_html(lock):

  1. #爬取文章详情页
  2. detail_url_list = variables.detail_url_list
  3. while True:
  5.     if len(variables.detail_url_list):
  6.         lock.acquire()
  7.         if len(detail_url_list):
  8.             url = detail_url_list.pop()
  9.             lock.release()
  10.             # for url in detail_url_list:
  11.             print("get detail html started")
  12.             time.sleep(2)
  13.             print("get detail html end")
  14.         else:
  15.             lock.release()
  16.             time.sleep(1)

def get_detail_url(lock):

  1. # 爬取文章列表页
  2. detail_url_list = variables.detail_url_list
  3. while True:
  4.     print("get detail url started")
  5.     time.sleep(4)
  6.     for i in range(20):
  7.         lock.acquire() # 获取锁
  8.         if len(detail_url_list) >= 10:
  9.             lock.release() #释放锁
  10.             time.sleep(1)
  11.         else:
  12.             detail_url_list.append("http://projectsedu.com/{id}".format(id=i))
  13.             lock.release()
  14.     print("get detail url end")

1. 线程通信方式- 共享变量

if name == “main“: lock = RLock() thread_detail_url = threading.Thread(target=get_detail_url, args=(lock,)) for i in range(10): html_thread = threading.Thread(target=get_detail_html, args=(lock,)) html_thread.start()

  1. # # thread2 = GetDetailUrl("get_detail_url")
  2. start_time = time.time()
  3. # thread_detail_url.start()
  4. # thread_detail_url1.start()
  5. #
  6. # thread1.join()
  7. # thread2.join()
  9. #当主线程退出的时候， 子线程kill掉
  10. print ("last time: {}".format(time.time()-start_time))
  2. <a name="eKpwg"></a>
  3. # threading 线程间通信 -queue
  4. queue是线程安全的不需要加锁
  5. ```python
  6. #通过queue的方式进行线程间同步
  7. from queue import Queue
  10. import time
  11. import threading
  14. def get_detail_html(queue):
  15.     #爬取文章详情页
  16.     while True:
  17.         url = queue.get()
  18.         # for url in detail_url_list:
  19.         print("get detail html started")
  20.         time.sleep(2)
  21.         print("get detail html end")
  24. def get_detail_url(queue):
  25.     # 爬取文章列表页
  26.     while True:
  27.         print("get detail url started")
  28.         time.sleep(4)
  29.         for i in range(20):
  30.             queue.put("http://projectsedu.com/{id}".format(id=i))
  31.         print("get detail url end")
  34. #1. 线程通信方式- 共享变量
  36. if  __name__ == "__main__":
  37.     detail_url_queue = Queue(maxsize=1000)  #超过数量会阻塞
  40.     thread_detail_url = threading.Thread(target=get_detail_url, args=(detail_url_queue,))
  41.     for i in range(10):
  42.         html_thread = threading.Thread(target=get_detail_html, args=(detail_url_queue,))
  43.         html_thread.start()
  44.     # # thread2 = GetDetailUrl("get_detail_url")
  45.     start_time = time.time()
  46.     # thread_detail_url.start()
  47.     # thread_detail_url1.start()
  48.     #
  49.     # thread1.join()
  50.     # thread2.join()
  51.     detail_url_queue.task_done()
  52.     detail_url_queue.join()
  54.     #当主线程退出的时候， 子线程kill掉
  55.     print ("last time: {}".format(time.time()-start_time))

复杂的线程同步 condition条件变量

  1. import threading
  3. #条件变量， 用于复杂的线程间同步
  4. 下面代码会一个线程执行完才能另一个线程执行
  5. # class XiaoAi(threading.Thread):
  6. #     def __init__(self, lock):
  7. #         super().__init__(name="小爱")
  8. #         self.lock = lock
  9. #
  10. #     def run(self):
  11. #         self.lock.acquire()
  12. #         print("{} : 在 ".format(self.name))
  13. #         self.lock.release()
  14. #
  15. #         self.lock.acquire()
  16. #         print("{} : 好啊 ".format(self.name))
  17. #         self.lock.release()
  18. #
  19. # class TianMao(threading.Thread):
  20. #     def __init__(self, lock):
  21. #         super().__init__(name="天猫精灵")
  22. #         self.lock = lock
  23. #
  24. #     def run(self):
  25. #
  26. #         self.lock.acquire()
  27. #         print("{} : 小爱同学 ".format(self.name))
  28. #         self.lock.release()
  29. #
  30. #         self.lock.acquire()
  31. #         print("{} : 我们来对古诗吧 ".format(self.name))
  32. #         self.lock.release()
  34. #通过condition完成协同读诗
  35. class XiaoAi(threading.Thread):
  36.     def __init__(self, cond):
  37.         super().__init__(name="小爱")
  38.         self.cond = cond
  40.     def run(self):
  41.         with self.cond:
  42.             self.cond.wait()
  43.             print("{} : 在 ".format(self.name))
  44.             self.cond.notify()   #通知另一个线程可以执行了
  46.             self.cond.wait()    #等待通知
  47.             print("{} : 好啊 ".format(self.name))
  48.             self.cond.notify()
  50.             self.cond.wait()
  51.             print("{} : 君住长江尾 ".format(self.name))
  52.             self.cond.notify()
  54.             self.cond.wait()
  55.             print("{} : 共饮长江水 ".format(self.name))
  56.             self.cond.notify()
  58.             self.cond.wait()
  59.             print("{} : 此恨何时已 ".format(self.name))
  60.             self.cond.notify()
  62.             self.cond.wait()
  63.             print("{} : 定不负相思意 ".format(self.name))
  64.             self.cond.notify()
  66. class TianMao(threading.Thread):
  67.     def __init__(self, cond):
  68.         super().__init__(name="天猫精灵")
  69.         self.cond = cond
  71.     def run(self):
  72.         with self.cond:
  73.             print("{} : 小爱同学 ".format(self.name))
  74.             self.cond.notify()
  75.             self.cond.wait()
  77.             print("{} : 我们来对古诗吧 ".format(self.name))
  78.             self.cond.notify()
  79.             self.cond.wait()
  81.             print("{} : 我住长江头 ".format(self.name))
  82.             self.cond.notify()
  83.             self.cond.wait()
  85.             print("{} : 日日思君不见君 ".format(self.name))
  86.             self.cond.notify()
  87.             self.cond.wait()
  89.             print("{} : 此水几时休 ".format(self.name))
  90.             self.cond.notify()
  91.             self.cond.wait()
  93.             print("{} : 只愿君心似我心 ".format(self.name))
  94.             self.cond.notify()
  95.             self.cond.wait()
  99. if __name__ == "__main__":
  100.     from concurrent import futures
  101.     cond = threading.Condition()
  102.     xiaoai = XiaoAi(cond)
  103.     tianmao = TianMao(cond)
  105.     #启动顺序很重要
  106.     #在调用with cond之后才能调用wait或者notify方法
  107.     #condition有两层锁， 一把底层锁会在线程调用了wait方法的时候释放， 上面的锁会在每次调用wait的时候分配一把并放入到cond的等待队列中，等到notify方法的唤醒
  108.     xiaoai.start()
  109.     tianmao.start()