gevent 高并发、 RabbitMQ 消息队列、Celery 分布式的案例和说明

1. gevent 高并发请求示例

gevent​​：基于协程的Python库，通过异步非阻塞模式实现高并发请求。例如，同时抓取100个网页时，无需等待每个请求完成，提升效率。

import gevent
from gevent import monkey
monkey.patch_all()  # 替换标准库的阻塞IO
import requestsdef fetch_url(url):try:response = requests.get(url, timeout=5)print(f"URL: {url} 状态码: {response.status_code} 长度: {len(response.text)}")except Exception as e:print(f"请求失败 {url}: {str(e)}")urls = ['https://www.baidu.com','https://www.qq.com','https://www.taobao.com',# 可添加更多URL
] * 25  # 重复25次达到100个请求# 创建协程池并发执行
jobs = [gevent.spawn(fetch_url, url) for url in urls]
gevent.joinall(jobs, timeout=10)

运行：

pip install gevent requests
python demo.py

---

2. RabbitMQ 消息队列示例

消息队列（MQ）​​：如RabbitMQ或Kafka，用于解耦任务生产与消费。例如，将爬虫任务拆分为多个子任务，通过队列分发给不同服务器执行，避免单点故障。

生产者（producer.py）

import pikaconnection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()# 声明持久化队列
channel.queue_declare(queue='task_queue', durable=True)# 发送10个测试任务
for i in range(10):message = f'任务数据 {i}'channel.basic_publish(exchange='',routing_key='task_queue',body=message,properties=pika.BasicProperties(delivery_mode=2,  # 消息持久化))print(f" [x] 已发送 {message}")connection.close()

消费者（consumer.py）

import pika
import timedef callback(ch, method, properties, body):print(f" [x] 收到 {body.decode()}")time.sleep(1)  # 模拟任务处理ch.basic_ack(delivery_tag=method.delivery_tag)  # 手动确认connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='task_queue', durable=True)
channel.basic_qos(prefetch_count=1)  # 公平分发
channel.basic_consume(queue='task_queue', on_message_callback=callback)print(' [*] 等待消息...')
channel.start_consuming()

运行：

pip install pika
# 先启动RabbitMQ服务
python producer.py  # 另一个终端运行
python consumer.py

---

3. Celery 分布式任务示例

​​Celery​​：Python的分布式任务队列框架，支持定时任务和异步执行。例如，定时抓取新闻网站，任务自动分配到多台机器运行。

创建任务文件（celery_demo.py）

from celery import Celeryapp = Celery('tasks',broker='redis://localhost:6379/0',backend='redis://localhost:6379/0'
)@app.task
def fetch_news(url):# 实际应使用requests等库print(f"正在抓取: {url}")return f"{url} 抓取完成"# 定时配置
app.conf.beat_schedule = {'every-10-seconds': {'task': 'celery_demo.fetch_news','schedule': 10.0,  # 每10秒执行'args': ('https://news.example.com',)},
}

运行：

pip install celery redis
# 启动Worker
celery -A celery_demo worker --loglevel=info
# 另一个终端启动定时任务
celery -A celery_demo beat --loglevel=info

---

关键点说明：

1. gevent：

   • 通过协程实现并发（非并行）
   • monkey.patch_all() 替换标准库的阻塞调用
   • 适合I/O密集型场景

2. RabbitMQ：

   • 使用持久化队列防止消息丢失
   • 手动消息确认保证可靠性
   • 公平分发（prefetch_count=1）

3. Celery：

   • 使用Redis作为消息代理和结果存储
   • 支持定时任务（需启动beat）
   • 可分布式部署多个Worker

实际生产环境中需考虑：

• 错误重试机制
• 日志记录
• 资源监控
• 集群部署配置