#当爬虫遇上 RabbitMQ：高效解耦的分布式爬取实践

import pika
import json

def init_mq_producer():
    """初始化生产者连接，声明持久化的待爬队列"""
    # 建立阻塞连接（本地测试用 BlockingConnection 足够，生产环境推荐 SelectConnection）
    connection = pika.BlockingConnection(pika.ConnectionParameters(host="localhost"))
    channel = connection.channel()

    # 声明队列：durable=True 表示队列持久化（MQ 重启后队列依然存在）
    channel.queue_declare(queue="scrape_url_queue", durable=True)
    return connection, channel

def send_scrape_task(channel, url: str, extra_params: dict = None):
    """发送单个爬取任务（将任务序列化为 JSON，方便传递复杂参数）"""
    task = {"url": url, "params": extra_params or {}}
    # delivery_mode=2 表示消息持久化到磁盘
    channel.basic_publish(
        exchange="",  # 使用默认直连交换机，routing_key 直接匹配队列名
        routing_key="scrape_url_queue",
        body=json.dumps(task, ensure_ascii=False),
        properties=pika.BasicProperties(delivery_mode=2),
    )
    print(f"✅ 生产者已发送任务：{url}")

if __name__ == "__main__":
    # 测试：丢 3 个百度新闻子域名的 URL
    conn, ch = init_mq_producer()
    test_urls = [
        "https://news.baidu.com/",
        "https://finance.baidu.com/",
        "https://tech.baidu.com/",
    ]
    for u in test_urls:
        send_scrape_task(ch, u)
    conn.close()

import pika
import json
import requests
from bs4 import BeautifulSoup

def init_mq_consumer():
    """初始化消费者连接，并开启公平分发"""
    connection = pika.BlockingConnection(pika.ConnectionParameters(host="localhost"))
    channel = connection.channel()

    # 队列声明必须与生产者一致，否则会因找不到队列而报错
    channel.queue_declare(queue="scrape_url_queue", durable=True)

    # 公平分发：每个消费者只预取 1 个任务，处理完再领取下一个
    channel.basic_qos(prefetch_count=1)
    return connection, channel

def scrape_page(task: dict):
    """模拟真实爬取：抓取页面标题，并返回结果字典"""
    try:
        resp = requests.get(task["url"], headers={"User-Agent": "Mozilla/5.0"}, timeout=5)
        resp.raise_for_status()  # 遇到 4xx/5xx 直接抛出异常
        soup = BeautifulSoup(resp.text, "html.parser")
        return {"url": task["url"], "title": soup.title.string.strip(), "status": "success"}
    except Exception as e:
        return {"url": task["url"], "error": str(e), "status": "failed"}

def task_callback(ch, method, properties, body):
    """回调函数：反序列化 → 爬取 → 手动确认 → 继续等待新任务"""
    task = json.loads(body.decode("utf-8"))
    print(f"🔄 消费者正在处理：{task['url']}")

    result = scrape_page(task)
    print(f"📄 处理结果：{result}")

    # 手动确认：告诉 MQ 这个任务已处理完，可以从队列中删除
    ch.basic_ack(delivery_tag=method.delivery_tag)

if __name__ == "__main__":
    conn, ch = init_mq_consumer()
    # 注册回调，开始消费
    ch.basic_consume(
        queue="scrape_url_queue",
        on_message_callback=task_callback,
        auto_ack=False,  # 必须关闭自动确认！否则消费者刚拿到任务 MQ 就会删掉
    )
    print("⏳ 消费者等待任务中... 按 CTRL+C 退出")
    try:
        ch.start_consuming()
    except KeyboardInterrupt:
        print("\n👋 消费者已停止")
        conn.close()

# 声明队列时设置最大优先级
channel.queue_declare(
    queue="priority_scrape_queue",
    durable=True,
    arguments={"x-max-priority": 10}   # 最大优先级 10，避免设太高影响性能
)

# 发送任务时携带优先级属性
def send_priority_task(channel, url, priority):
    ch.basic_publish(
        exchange="",
        routing_key="priority_scrape_queue",
        body=json.dumps({"url": url}),
        properties=pika.BasicProperties(delivery_mode=2, priority=priority),
    )

# 示例：先发历史归档（优先级 1），再发热榜新闻（优先级 10）
send_priority_task(ch, "https://news.baidu.com/history/20240101", 1)
send_priority_task(ch, "https://news.baidu.com/", 10)
# 消费者将优先收到优先级为 10 的热榜任务