假设在传统的生产者/消费者模式中,生产者将生产两种类型的元素,A和B。缓冲队列应实现以下逻辑:
A
B
为简单起见,假设队列的大小是无限的。不使用2个队列是否可以实现此功能?
ne5o7dgx1#
你可能可以子类化一个collections.deque并得到类似这样的结果:
collections.deque
from collection import deque class Foo(deque[str]): def append(self, val: str): if val == 'A': super().append(val) else: try: while True: super().remove('B') except ValueError: pass super().append(val)
from collection import deque
class Foo(deque[str]):
def append(self, val: str):
if val == 'A':
super().append(val)
else:
try:
while True:
super().remove('B')
except ValueError:
pass
字符串
while True
q = Foo(['B', 'B', 'B', 'B'])
__init __
'B'
然后你可以继续把任务放在队列中:
tasks = ['A', 'A', 'A', 'B', 'A', 'B', 'A', 'B', 'A', 'A']q = Foo()for i in tasks: q.append(i) print(q)Out:Foo(['A'])Foo(['A', 'A'])Foo(['A', 'A', 'A'])Foo(['A', 'A', 'A', 'B'])Foo(['A', 'A', 'A', 'B', 'A'])Foo(['A', 'A', 'A', 'A', 'B'])Foo(['A', 'A', 'A', 'A', 'B', 'A'])Foo(['A', 'A', 'A', 'A', 'A', 'B'])Foo(['A', 'A', 'A', 'A', 'A', 'B', 'A'])Foo(['A', 'A', 'A', 'A', 'A', 'B', 'A', 'A'])
tasks = ['A', 'A', 'A', 'B', 'A', 'B', 'A', 'B', 'A', 'A']
q = Foo()
for i in tasks:
q.append(i)
print(q)
Out:
Foo(['A'])
Foo(['A', 'A'])
Foo(['A', 'A', 'A'])
Foo(['A', 'A', 'A', 'B'])
Foo(['A', 'A', 'A', 'B', 'A'])
Foo(['A', 'A', 'A', 'A', 'B'])
Foo(['A', 'A', 'A', 'A', 'B', 'A'])
Foo(['A', 'A', 'A', 'A', 'A', 'B'])
Foo(['A', 'A', 'A', 'A', 'A', 'B', 'A'])
Foo(['A', 'A', 'A', 'A', 'A', 'B', 'A', 'A'])
型最后,使用popleft()以FIFO顺序获取元素。deque也应该是线程安全的:https://docs.python.org/3/library/collections.html#collections.deque
popleft()
deque
gpfsuwkq2#
与其试图从队列中驱逐B,不如在弹出B时检查它是否是队列中唯一的一个,这可能是最简单的方法。
queue = deque()def push(x): queue.append(x)def pop(x): while True: result = queue.popleft() if result == 'B' and 'B' in queue: continue return result
queue = deque()
def push(x):
queue.append(x)
def pop(x):
result = queue.popleft()
if result == 'B' and 'B' in queue:
continue
return result
字符串再多做一点努力,你就可以统计队列中B的数量,只在计数下降到0时才处理B。这将保存'B' in queue的成本,但计数必须通过一个锁来访问才是线程安全的。
'B' in queue
ddarikpa3#
从潜在的长任务队列中踢出现有的不重要任务的更有效的方法是使用双向链表作为队列,并维护对不重要任务的节点的引用(如果有的话),以便可以在 O(1) 时间复杂度内删除引用的节点。使用llist.dllist,一个来自llist模块的双向链表的优秀实现:
llist.dllist
llist
from llist import dllistfrom dataclasses import dataclass@dataclassclass Task: name: strclass UnimportantTask(Task): passclass Tasks: def __init__(self): self.queue = dllist() self.unimportant_task = None def add(self, task): node = self.queue.appendright(task) if isinstance(task, UnimportantTask): if self.unimportant_task: self.queue.remove(self.unimportant_task) self.unimportant_task = node def next(self): if self.queue: return self.queue.popleft()
from llist import dllist
from dataclasses import dataclass
@dataclass
class Task:
name: str
class UnimportantTask(Task):
class Tasks:
def __init__(self):
self.queue = dllist()
self.unimportant_task = None
def add(self, task):
node = self.queue.appendright(task)
if isinstance(task, UnimportantTask):
if self.unimportant_task:
self.queue.remove(self.unimportant_task)
self.unimportant_task = node
def next(self):
if self.queue:
return self.queue.popleft()
字符串以便:
tasks = Tasks()tasks.add(Task('A1'))tasks.add(Task('A2'))tasks.add(UnimportantTask('B1'))tasks.add(Task('A3'))tasks.add(UnimportantTask('B2'))tasks.add(UnimportantTask('B3'))tasks.add(Task('A4'))while task := tasks.next(): print(task)
tasks = Tasks()
tasks.add(Task('A1'))
tasks.add(Task('A2'))
tasks.add(UnimportantTask('B1'))
tasks.add(Task('A3'))
tasks.add(UnimportantTask('B2'))
tasks.add(UnimportantTask('B3'))
tasks.add(Task('A4'))
while task := tasks.next():
print(task)
型产出:
Task(name='A1')Task(name='A2')Task(name='A3')UnimportantTask(name='B3')Task(name='A4')
Task(name='A1')
Task(name='A2')
Task(name='A3')
UnimportantTask(name='B3')
Task(name='A4')
型演示:https://replit.com/@blhsing1/UnimportantTaskQueue#main.py
3条答案
按热度按时间ne5o7dgx1#
你可能可以子类化一个
collections.deque
并得到类似这样的结果:字符串
while True
块在这里,以防你从一个可迭代对象初始化队列,并且有超过1个“B”来开始,因为没有什么能阻止你这样做:q = Foo(['B', 'B', 'B', 'B'])
。如果你想,你也可以ovverride父对象的__init __
,以确保队列中总是只有1个'B'
。然后你可以继续把任务放在队列中:
型
最后,使用
popleft()
以FIFO顺序获取元素。deque
也应该是线程安全的:https://docs.python.org/3/library/collections.html#collections.dequegpfsuwkq2#
与其试图从队列中驱逐B,不如在弹出B时检查它是否是队列中唯一的一个,这可能是最简单的方法。
字符串
再多做一点努力,你就可以统计队列中B的数量,只在计数下降到0时才处理B。这将保存
'B' in queue
的成本,但计数必须通过一个锁来访问才是线程安全的。ddarikpa3#
从潜在的长任务队列中踢出现有的不重要任务的更有效的方法是使用双向链表作为队列,并维护对不重要任务的节点的引用(如果有的话),以便可以在 O(1) 时间复杂度内删除引用的节点。
使用
llist.dllist
,一个来自llist
模块的双向链表的优秀实现:字符串
以便:
型
产出:
型
演示:https://replit.com/@blhsing1/UnimportantTaskQueue#main.py