#当爬虫遇上 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):
    """发送单个爬取任务（序列化后发，方便传复杂参数）"""
    task = {"url": url, "params": extra_params or {}}
    # 持久化任务消息：delivery_mode=2 表示消息写入磁盘
    channel.basic_publish(
        exchange="",  # 空 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):
    """模拟真实爬虫：取title，避免写死，方便后续扩展"""
    try:
        resp = requests.get(task["url"], headers={"User-Agent": "Mozilla/5.0"}, timeout=5)
        resp.raise_for_status()  # 抛出HTTP错误（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，别设太大影响性能
)

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

# 测试：先发历史（优先级1），再发热榜（优先级10），但消费者会先拿到热榜
send_priority_task(ch, "https://news.baidu.com/history/20240101", 1)
send_priority_task(ch, "https://news.baidu.com/", 10)