目录

原理

实现代码

示例

---

原理

消息队列是一个先进先出栈，每次都处理第一项，处理完了过后会删除这个消息，这是一个简单的消息队列图：

实现代码

首先消息队列需要一个队列，我们用Python里的列表：

self.queue = []

接着我们要实现把一个消息加入队列和处理消息：

def push(self, text):self.queue.append(text)def pull(self):test = self.queue[0]self.queue.pop()return test

示例

然后我们来验证一下：

import time'''
消息队列的类
'''mq = MessageQueue()
for i in range(1, 21):mq.push(f"[Error {i}] author is not find")print(mq.pull(), end="")time.sleep(0.1)print(".", end="")time.sleep(0.1)print(".", end="")time.sleep(0.1)print(".")

结果：

[Error 1] author is not find...
[Error 2] author is not find...
[Error 3] author is not find...
[Error 4] author is not find...
[Error 5] author is not find...
[Error 6] author is not find...
[Error 7] author is not find...
[Error 8] author is not find...
[Error 9] author is not find...
[Error 10] author is not find...
[Error 11] author is not find...
[Error 12] author is not find...
[Error 13] author is not find...
[Error 14] author is not find...
[Error 15] author is not find...
[Error 16] author is not find...
[Error 17] author is not find...
[Error 18] author is not find...
[Error 19] author is not find...
[Error 20] author is not find...

Process finished with exit code 0

列表：

[] 

完整代码

import timeclass MessageQueue(object):def __init__(self):self.queue = []def push(self, text):self.queue.append(text)def pull(self):test = self.queue[0]self.queue.pop()return testmq = MessageQueue()
for i in range(1, 21):mq.push(f"[Error {i}] author is not find")print(mq.pull(), end="")time.sleep(0.1)print(".", end="")time.sleep(0.1)print(".", end="")time.sleep(0.1)print(".")