如何通過Python實現(xiàn)一個消息隊列

更新時間：2025年02月21日 15:24:40 作者：升訊威在線客服系統(tǒng)

這篇文章主要為大家詳細介紹了如何通過Python實現(xiàn)一個簡單的消息隊列,文中的示例代碼講解詳細,感興趣的小伙伴可以跟隨小編一起學(xué)習(xí)一下

如何通過 python 實現(xiàn)消息隊列

1. 使用 Python 內(nèi)置的 queue.Queue（適用于單機應(yīng)用）

queue.Queue 提供了線程安全的隊列操作，適合在多線程應(yīng)用中使用。

import queue
import threading
import time

# 創(chuàng)建一個先進先出（FIFO）隊列
msg_queue = queue.Queue()

# 生產(chǎn)者線程
def producer():
    for i in range(5):
        time.sleep(1)  # 模擬一些處理
        msg = f"消息{i}"
        msg_queue.put(msg)  # 將消息放入隊列
        print(f"生產(chǎn)者放入：{msg}")

# 消費者線程
def consumer():
    while True:
        msg = msg_queue.get()  # 從隊列獲取消息
        if msg is None:  # 終止條件
            break
        print(f"消費者處理：{msg}")
        msg_queue.task_done()  # 標記任務(wù)已完成

# 創(chuàng)建生產(chǎn)者和消費者線程
producer_thread = threading.Thread(target=producer)
consumer_thread = threading.Thread(target=consumer)

# 啟動線程
producer_thread.start()
consumer_thread.start()

# 等待生產(chǎn)者線程完成
producer_thread.join()

# 向消費者線程發(fā)送終止信號
msg_queue.put(None)

# 等待消費者線程完成
consumer_thread.join()

2. 使用 Redis（適用于分布式應(yīng)用）

Redis 是一個高效的內(nèi)存數(shù)據(jù)存儲，可以用作分布式消息隊列。你可以使用 redis-py 庫與 Redis 進行交互。

pip install redis

import redis
import time

# 創(chuàng)建 Redis 連接
r = redis.StrictRedis(host='localhost', port=6379, db=0)

# 生產(chǎn)者：將消息放入隊列
def producer():
    for i in range(5):
        time.sleep(1)  # 模擬一些處理
        msg = f"消息{i}"
        r.lpush('msg_queue', msg)  # 將消息推送到隊列
        print(f"生產(chǎn)者放入：{msg}")

# 消費者：從隊列中獲取消息
def consumer():
    while True:
        msg = r.brpop('msg_queue')[1].decode('utf-8')  # 從隊列中獲取消息
        print(f"消費者處理：{msg}")

# 啟動生產(chǎn)者和消費者
producer()
consumer()

3. 使用 RabbitMQ（適用于更復(fù)雜的消息傳遞）

RabbitMQ 是一個強大的消息代理系統(tǒng)，支持多種消息隊列模式。如果需要高度可靠和高性能的消息隊列，可以使用 pika 庫連接 RabbitMQ。

pip install pika

import pika

# 連接到 RabbitMQ
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# 聲明一個隊列
channel.queue_declare(queue='msg_queue')

# 生產(chǎn)者：發(fā)送消息
def producer():
    for i in range(5):
        msg = f"消息{i}"
        channel.basic_publish(exchange='',
                              routing_key='msg_queue',
                              body=msg)
        print(f"生產(chǎn)者發(fā)送：{msg}")

# 消費者：接收并處理消息
def consumer(ch, method, properties, body):
    print(f"消費者處理：{body.decode('utf-8')}")

# 啟動消費者
channel.basic_consume(queue='msg_queue', on_message_callback=consumer, auto_ack=True)

# 啟動生產(chǎn)者和消費者
producer()
print('等待消息...')
channel.start_consuming()

選擇合適的實現(xiàn)

如果你是單機應(yīng)用并且需要線程安全的隊列，使用 queue.Queue。
如果你的應(yīng)用是分布式的，使用 Redis 或 RabbitMQ 更加合適，它們提供了高可用性、消息持久化和可靠的消息傳遞機制。

如何把 http 請求放在隊列中執(zhí)行

將 HTTP 請求放入隊列并異步執(zhí)行的場景通常用于處理高并發(fā)、后臺任務(wù)、延遲任務(wù)等情況。你可以使用消息隊列系統(tǒng)（如 queue.Queue、Redis 或 RabbitMQ）來將 HTTP 請求放入隊列，消費隊列中的任務(wù)并執(zhí)行相應(yīng)的 HTTP 請求。

這里我會展示幾種不同的實現(xiàn)方式，供你參考。

1. 使用 queue.Queue 和 requests 庫

你可以將 HTTP 請求封裝為任務(wù)，并將其放入隊列中，然后使用多個消費者線程異步處理隊列中的請求。

import queue
import threading
import time
import requests

# 創(chuàng)建一個隊列
task_queue = queue.Queue()

# HTTP 請求任務(wù)處理函數(shù)
def handle_request():
    while True:
        url = task_queue.get()  # 從隊列中獲取任務(wù)
        if url is None:  # 終止條件
            break
        try:
            response = requests.get(url)
            print(f"請求 {url} 的響應(yīng)狀態(tài): {response.status_code}")
        except Exception as e:
            print(f"請求 {url} 失敗: {e}")
        task_queue.task_done()  # 標記任務(wù)完成

# 生產(chǎn)者：將 HTTP 請求放入隊列
def producer():
    urls = [
        "https://jsonplaceholder.typicode.com/posts/1",
        "https://jsonplaceholder.typicode.com/posts/2",
        "https://jsonplaceholder.typicode.com/posts/3"
    ]
    
    for url in urls:
        print(f"將 URL {url} 放入隊列")
        task_queue.put(url)
        time.sleep(1)  # 模擬任務(wù)產(chǎn)生的延遲

# 創(chuàng)建多個消費者線程
consumer_threads = []
for i in range(3):
    t = threading.Thread(target=handle_request)
    t.start()
    consumer_threads.append(t)

# 啟動生產(chǎn)者線程
producer_thread = threading.Thread(target=producer)
producer_thread.start()

# 等待生產(chǎn)者線程完成
producer_thread.join()

# 向消費者線程發(fā)送終止信號
for _ in range(3):
    task_queue.put(None)

# 等待消費者線程完成
for t in consumer_threads:
    t.join()

2. 使用 Redis 和 requests 庫

Redis 可以作為一個分布式的消息隊列，適用于分布式系統(tǒng)中將 HTTP 請求放入隊列并異步執(zhí)行。你可以使用 Redis 的列表數(shù)據(jù)結(jié)構(gòu)（lpush、brpop）來實現(xiàn)。

import redis
import requests
import time

# 創(chuàng)建 Redis 連接
r = redis.StrictRedis(host='localhost', port=6379, db=0)

# 生產(chǎn)者：將 HTTP 請求放入隊列
def producer():
    urls = [
        "https://jsonplaceholder.typicode.com/posts/1",
        "https://jsonplaceholder.typicode.com/posts/2",
        "https://jsonplaceholder.typicode.com/posts/3"
    ]
    
    for url in urls:
        print(f"將 URL {url} 放入 Redis 隊列")
        r.lpush('task_queue', url)
        time.sleep(1)  # 模擬任務(wù)產(chǎn)生的延遲

# 消費者：從隊列中獲取請求并執(zhí)行
def consumer():
    while True:
        url = r.brpop('task_queue')[1].decode('utf-8')  # 從隊列中獲取任務(wù)
        try:
            response = requests.get(url)
            print(f"請求 {url} 的響應(yīng)狀態(tài): {response.status_code}")
        except Exception as e:
            print(f"請求 {url} 失敗: {e}")

# 啟動生產(chǎn)者和消費者
producer_thread = threading.Thread(target=producer)
consumer_thread = threading.Thread(target=consumer)

producer_thread.start()
consumer_thread.start()

# 等待生產(chǎn)者線程完成
producer_thread.join()

# 由于 Redis 隊列會一直阻塞等待任務(wù)，可以根據(jù)需要添加退出邏輯

3. 使用 RabbitMQ 和 requests 庫

RabbitMQ 提供了強大的消息隊列機制，適合用于大規(guī)模的消息傳遞。你可以創(chuàng)建一個任務(wù)隊列，將 HTTP 請求放入隊列中，并通過消費者處理隊列中的請求。

import pika
import requests
import time

# 連接到 RabbitMQ
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# 聲明隊列
channel.queue_declare(queue='http_requests')

# 生產(chǎn)者：將 HTTP 請求放入隊列
def producer():
    urls = [
        "https://jsonplaceholder.typicode.com/posts/1",
        "https://jsonplaceholder.typicode.com/posts/2",
        "https://jsonplaceholder.typicode.com/posts/3"
    ]
    
    for url in urls:
        print(f"將 URL {url} 放入 RabbitMQ 隊列")
        channel.basic_publish(exchange='',
                              routing_key='http_requests',
                              body=url)
        time.sleep(1)  # 模擬任務(wù)產(chǎn)生的延遲

# 消費者：處理 HTTP 請求
def consumer(ch, method, properties, body):
    url = body.decode('utf-8')
    try:
        response = requests.get(url)
        print(f"請求 {url} 的響應(yīng)狀態(tài): {response.status_code}")
    except Exception as e:
        print(f"請求 {url} 失敗: {e}")

# 啟動消費者
channel.basic_consume(queue='http_requests', on_message_callback=consumer, auto_ack=True)

# 啟動生產(chǎn)者
producer_thread = threading.Thread(target=producer)
producer_thread.start()

# 啟動消費者并等待消息
print('等待消費者處理 HTTP 請求...')
producer_thread.join()
channel.start_consuming()

4. 使用 Celery 異步任務(wù)隊列

Celery 是一個強大的異步任務(wù)隊列，適用于分布式任務(wù)執(zhí)行。通過 Celery，你可以把 HTTP 請求封裝為任務(wù)，放入隊列中進行異步執(zhí)行。

首先，你需要安裝 Celery 和 requests：

pip install celery requests

然后在 celery.py 中配置 Celery：

from celery import Celery
import requests

app = Celery('http_requests', broker='redis://localhost:6379/0')

@app.task
def fetch_url(url):
    try:
        response = requests.get(url)
        print(f"請求 {url} 的響應(yīng)狀態(tài): {response.status_code}")
    except Exception as e:
        print(f"請求 {url} 失敗: {e}")

然后在主程序中提交任務(wù)：

from celery import Celery
from celery.py import fetch_url

# 添加任務(wù)到隊列
fetch_url.apply_async(args=["https://jsonplaceholder.typicode.com/posts/1"])
fetch_url.apply_async(args=["https://jsonplaceholder.typicode.com/posts/2"])
fetch_url.apply_async(args=["https://jsonplaceholder.typicode.com/posts/3"])

啟動 Celery Worker：