Queue(队列对象)
Queue 是 python 中的标准库, 可以直接 import Queue 引用; 队列是线程间最常用的交换数据的形式
python 下多线程的思考
对于资源, 加锁是个重要的环节. 因为 python 原生的 list,dict 等, 都是 not thread safe 的. 而 Queue, 是线程安全的, 因此在满足使用条件下, 建议使用队列
初始化: class Queue.Queue(maxsize) FIFO 先进先出
包中的常用方法:
Queue.qsize() 返回队列的大小
Queue.empty() 如果队列为空, 返回 True, 反之 False
Queue.full() 如果队列满了, 返回 True, 反之 False
Queue.full 与 maxsize 大小对应
Queue.get([block[, timeout]])获取队列, timeout 等待时间
创建一个 "队列" 对象
- import Queue
- myqueue = Queue.Queue(maxsize = 10)
将一个值放入队列中
myqueue.put(10)
将一个值从队列中取出
- myqueue.get()
- # -*- coding:utf-8 -*-
- # 使用了线程库
- import threading
- # 队列
- from Queue import Queue
- # 解析库
- from lxml import etree
- # 请求处理
- import requests
- # json 处理
- import json
- import time
- class ThreadCrawl(threading.Thread):
- def __init__(self, threadName, pageQueue, dataQueue):
- #threading.Thread.__init__(self)
- # 调用父类初始化方法
- super(ThreadCrawl, self).__init__()
- # 线程名
- self.threadName = threadName
- # 页码队列
- self.pageQueue = pageQueue
- # 数据队列
- self.dataQueue = dataQueue
- # 请求报头
- self.headers = {"User-Agent" : "Mozilla/5.0 (compatible; MSIE 9.0; Windows NT 6.1; Trident/5.0;"}
- def run(self):
- print "启动" + self.threadName
- while not CRAWL_EXIT:
- try:
- # 取出一个数字, 先进先出
- # 可选参数 block, 默认值为 True
- #1. 如果对列为空, block 为 True 的话, 不会结束, 会进入阻塞状态, 直到队列有新的数据
- #2. 如果队列为空, block 为 False 的话, 就弹出一个 Queue.empty()异常,
- page = self.pageQueue.get(False)
- url = "http://www.qiushibaike.com/8hr/page/" + str(page) +"/"
- #print url
- content = requests.get(url, headers = self.headers).text
- time.sleep(1)
- self.dataQueue.put(content)
- #print len(content)
- except:
- pass
- print "结束" + self.threadName
- class ThreadParse(threading.Thread):
- def __init__(self, threadName, dataQueue, filename, lock):
- super(ThreadParse, self).__init__()
- # 线程名
- self.threadName = threadName
- # 数据队列
- self.dataQueue = dataQueue
- # 保存解析后数据的文件名
- self.filename = filename
- # 锁
- self.lock = lock
- def run(self):
- print "启动" + self.threadName
- while not PARSE_EXIT:
- try:
- html = self.dataQueue.get(False)
- self.parse(html)
- except:
- pass
- print "退出" + self.threadName
- def parse(self, html):
- # 解析为 HTML DOM
- html = etree.HTML(html)
- node_list = html.xpath('//div[contains(@id,"qiushi_tag")]')
- for node in node_list:
- # xpath 返回的列表, 这个列表就这一个参数, 用索引方式取出来, 用户名
- username = node.xpath('.//img/@alt')[0]
- # 图片连接
- image = node.xpath('.//div[@class="thumb"]//@src')#[0]
- # 取出标签下的内容, 段子内容
- content = node.xpath('.//div[@class="content"]/span')[0].text
- # 取出标签里包含的内容, 点赞
- zan = node.xpath('.//i')[0].text
- # 评论
- comments = node.xpath('.//i')[1].text
- items = {
- "username" : username,
- "image" : image,
- "content" : content,
- "zan" : zan,
- "comments" : comments
- }
- # with 后面有两个必须执行的操作:__enter__ 和 _exit__
- # 不管里面的操作结果如何, 都会执行打开, 关闭
- # 打开锁, 处理内容, 释放锁
- with self.lock:
- # 写入存储的解析后的数据
- self.filename.write(json.dumps(items, ensure_ascii = False).encode("utf-8") + "\n")
- CRAWL_EXIT = False
- PARSE_EXIT = False
- def main():
- # 页码的队列, 表示 20 个页面
- pageQueue = Queue(20)
- # 放入 1~10 的数字, 先进先出
- for i in range(1, 21):
- pageQueue.put(i)
- # 采集结果 (每页的 HTML 源码) 的数据队列, 参数为空表示不限制
- dataQueue = Queue()
- filename = open("duanzi.json", "a")
- # 创建锁
- lock = threading.Lock()
- # 三个采集线程的名字
- crawlList = ["采集线程 1 号", "采集线程 2 号", "采集线程 3 号"]
- # 存储三个采集线程的列表集合
- threadcrawl = []
- for threadName in crawlList:
- thread = ThreadCrawl(threadName, pageQueue, dataQueue)
- thread.start()
- threadcrawl.append(thread)
- # 三个解析线程的名字
- parseList = ["解析线程 1 号","解析线程 2 号","解析线程 3 号"]
- # 存储三个解析线程
- threadparse = []
- for threadName in parseList:
- thread = ThreadParse(threadName, dataQueue, filename, lock)
- thread.start()
- threadparse.append(thread)
- # 等待 pageQueue 队列为空, 也就是等待之前的操作执行完毕
- while not pageQueue.empty():
- pass
- # 如果 pageQueue 为空, 采集线程退出循环
- global CRAWL_EXIT
- CRAWL_EXIT = True
- print "pageQueue 为空"
- for thread in threadcrawl:
- thread.join()
- print "1"
- while not dataQueue.empty():
- pass
- global PARSE_EXIT
- PARSE_EXIT = True
- for thread in threadparse:
- thread.join()
- print "2"
- with lock:
- # 关闭文件
- filename.close()
- print "谢谢使用!"
- if __name__ == "__main__":
- main()
来源: http://www.bubuko.com/infodetail-2649989.html