博客
关于我
强烈建议你试试无所不能的chatGPT,快点击我
Basic FIFO Queue
阅读量:6035 次
发布时间:2019-06-20

本文共 2479 字,大约阅读时间需要 8 分钟。

Queue - 一种线程安全的FIFO实现

Python的Queue模块提供一种适用于多线程编程的FIFO实现。它可用于在生产者(producer)和消费者(consumer)之间线程安全(thread-safe)地传递消息或其它数据,因此多个线程可以共用同一个Queue实例。Queue的大小(元素的个数)可用来限制内存的使用。

Basic FIFO Queue

Queue类实现了一个基本的先进先出(FIFO)容器,使用put()将元素添加到序列尾端,get()从队列尾部移除元素。

+

 

from queue import Queueq = Queue()for i in range(3): q.put(i) while not q.empty(): print(q.get())

上例使用单线程演示了元素以插入顺序从队列中移除。结果如下:

0123

LIFO Queue

与标准FIFO实现Queue不同的是,LifoQueue使用后进先出序(会关联一个栈数据结构)。

from queue import LifoQueueq = LifoQueue()for i in range(3): q.put(i) while not q.empty(): print(q.get())

最后put()到队列的元素最先被get()

210

Priority Queue(优先队列)

除了按元素入列顺序外,有时需要根据队列中元素的特性来决定元素的处理顺序。例如,财务部门的打印任务可能比码农的代码打印任务优先级更高。PriorityQueue依据队列中内容的排序顺序(sort order)来决定那个元素将被检索。

from queue import PriorityQueueclass Job(object): def __init__(self, priority, description): self.priority = priority self.description = description print('New job:', description) return def __lt__(self, other): return self.priority < other.priority q = PriorityQueue() q.put(Job(5, 'Mid-level job')) q.put(Job(10, 'Low-level job')) q.put(Job(1, 'Important job')) while not q.empty(): next_job = q.get() print('Processing job', next_job.description)

在这个单线程示例中,job会严格按照优先级从队列中取出。如有有多个线程同时消耗这些job,在get()被调用时,job会依据其优先级被处理。

New job: Mid-level jobNew job: Low-level jobNew job: Important jobProcessing job: Important jobProcessing job: Mid-level jobProcessing job: Low-level job

Using Queues with Threads

下例通过创建一个简单的播客客户端来展示如何将Queue类和多线程结合使用。这个客户端会从一个或多个RSS源读取内容,先创建一个用于存放下载内容的队列,然后使用多线程并行地处理多个下载任务。

import timefrom queue import Queuefrom threading import Thread #: 自己写的解析模块 import feedparser num_fetch_threads = 2 enclosure_queue = Queue() feed_urls = ['http:xxx/xxx',] def downloadEnclosures(i, q): """ 线程worker函数 用于处理队列中的元素项,这些守护线程在一个无限循环中,只有当主线程结束时才会结束循环 """ while True: print('%s: Looking for the next enclosure' % i) url = q.get() print('%s: Downloading: %s' % (i, url)) #: 用sleep代替真实的下载 time.sleep(i + 2) q.task_done() for i in range(num_fetch_threads): worker = Thread(target=downloadEnclosures, args=(i, enclosure_queue)) worker.setDaemon(True) worker.start() for url in feed_urls: response = feedparser.parse(url, agent='fetch_podcasts.py') for entry in response['entries']: for enclosure in entry.get('enclosures', []): print('Queuing:', enclosure['url']) enclosure_queue.put(enclosure['url']) # Now wait for the queue to be empty, indicating that we have # processed all of the downloads. print('*** Main thread waiting') enclosure_queue.join() print('*** Done')

转载地址:http://mzlhx.baihongyu.com/

你可能感兴趣的文章
mysql通过配置文件进行优化
查看>>
省级网站群建设关注点
查看>>
工作第四天之采集资源
查看>>
innobackupex 在增量的基础上增量备份
查看>>
Windows Server 2012 R2 DirectAccess功能测试(2)App1服务器安装及配置
查看>>
基于清单的启动器的实现
查看>>
外网用户通过citrix打印慢的解决方法
查看>>
STL容器的使用
查看>>
关于std::map
查看>>
JXL导出Excel文件兼容性问题
查看>>
VBoot1.0发布,Vue & SpringBoot 综合开发入门
查看>>
centos7 安装wps 后 演示无法启动
查看>>
git简单命令
查看>>
LAMP编译部署
查看>>
XenDesktop7.6安装部署入门教程
查看>>
HashMap的工作原理及HashMap和Hashtable的区别
查看>>
GregorianCalendar日历程序
查看>>
Sublime 中运行 Shell 、Python、Lua、Groovy...等各种脚本
查看>>
【Java集合源码剖析】ArrayList源码剖析
查看>>
linux的目录结构
查看>>