一、业务背景
爬虫系统最怕的是无声无息地挂了——你不知道它什么时候停止工作了，等发现的时候已经几个小时没有新数据了。
我最早用ELK（Elasticsearch + Logstash + Kibana）做日志监控，但ELK太重了——三个组件都要部署和维护，对于中小规模的爬虫系统来说有点大材小用。
二、轻量级日志收集方案
我设计了一个更轻量的方案：
python
import loggingimport jsonimport redisimport timeclass StructuredLogger: """结构化日志记录器""" def init(self): self.redis = redis.Redis(decode_responses=True) self.log_key = "crawler:logs" def log(self, level, message, kwargs): log_entry = { 'timestamp': time.time(), 'level': level, 'message': message, kwargs } # 写入Redis List（最新的在最前面） self.redis.lpush(self.log_key, json.dumps(log_entry)) # 只保留最近10000条 self.redis.ltrim(self.log_key, 0, 9999) # 同时输出到控制台（开发环境） print(f"[{level}] {message}")# 使用logger = StructuredLogger()logger.log('INFO', '爬虫启动', spider='rakuten', worker_id=1)logger.log('ERROR', '请求失败', url='https://api.rakuten.co.jp', status_code=500)
三、关键指标监控
除了日志，还需要监控关键业务指标：
python
class MetricsCollector: def init(self): self.redis = redis.Redis(decode_responses=True) def record_request(self, platform, success=True): """记录请求""" key = f"metrics:requests:{platform}:{time.strftime('%Y%m%d%H')}" self.redis.hincrby(key, 'total', 1) if not success: self.redis.hincrby(key, 'failed', 1) self.redis.expire(key, 86400 7) # 保留7天 def record_item(self, platform, count=1): """记录采集到的商品数""" key = f"metrics:items:{platform}:{time.strftime('%Y%m%d%H')}" self.redis.hincrby(key, 'count', count) self.redis.expire(key, 86400 7) def get_recent_metrics(self, platform, hours=24): """获取最近N小时的指标""" metrics = [] for i in range(hours): hour = (time.time() - i 3600) hour_str = time.strftime('%Y%m%d%H', time.localtime(hour)) key = f"metrics:requests:{platform}:{hour_str}" data = self.redis.hgetall(key) if data: metrics.append({ 'hour': hour_str, 'total': int(data.get('total', 0)), 'failed': int(data.get('failed', 0)) }) return metrics
四、告警规则引擎
基于收集的指标，配置告警规则：
python
class AlertEngine: def init(self, metrics_collector): self.metrics = metrics_collector self.rules = [] self.alerted = {} # 记录已触发的告警，防止重复 def add_rule(self, rule): """添加告警规则""" self.rules.append(rule) def check(self): """定期检查所有规则""" for rule in self.rules: if self._evaluate_rule(rule): self._send_alert(rule) def _evaluate_rule(self, rule): """评估规则是否触发""" # 规则示例：{"platform": "rakuten", "metric": "success_rate", "threshold": 0.8, "window": 60} metrics = self.metrics.get_recent_metrics( rule['platform'], hours=rule.get('window', 1) ) total = sum(m['total'] for m in metrics) failed = sum(m['failed'] for m in metrics) if total == 0: # 没有请求，可能爬虫已停止 return rule.get('alert_on_zero', False) success_rate = (total - failed) / total return success_rate < rule['threshold']
五、告警通知
python
class AlertNotifier: def init(self): self.webhook_url = os.getenv('DINGTALK_WEBHOOK') def send(self, rule, metrics): message = f""" 【爬虫告警】 平台: {rule['platform']} 规则: {rule['description']} 当前成功率: {metrics['success_rate']:.2%} 阈值: {rule['threshold']:.2%} 时间: {time.strftime('%Y-%m-%d %H:%M:%S')} """ # 发送到钉钉/飞书/企业微信 requests.post(self.webhook_url, json={'text': message})
六、总结
轻量级监控方案的核心是：结构化日志 + 关键指标 + 告警规则。这套方案不需要部署ELK这样重量的组件，用Redis就能搞定，适合中小规模的爬虫系统。
第17篇
RabbitMQ在分布式订单系统中的三种交换器使用场景
一、业务背景
在跨境电商订单系统中，不同模块之间需要异步通信。RabbitMQ提供了四种交换器（Exchange）类型，每种适用于不同的场景。
二、Direct Exchange：精准路由
Direct Exchange根据Routing Key精确匹配队列。
python
import pikaclass DirectExchangeExample: def init(self): self.connection = pika.BlockingConnection( pika.ConnectionParameters('localhost') ) self.channel = self.connection.channel() # 声明Direct Exchange self.channel.exchange_declare( exchange='order_direct', exchange_type='direct' ) def setup_queues(self): # 不同优先级的订单走不同队列 self.channel.queue_declare('order_high_priority') self.channel.queue_declare('order_normal_priority') self.channel.queue_declare('order_low_priority') # 绑定 self.channel.queue_bind( exchange='order_direct', queue='order_high_priority', routing_key='high' ) self.channel.queue_bind( exchange='order_direct', queue='order_normal_priority', routing_key='normal' ) self.channel.queue_bind( exchange='order_direct', queue='order_low_priority', routing_key='low' ) def publish_order(self, order, priority='normal'): self.channel.basic_publish( exchange='order_direct', routing_key=priority, body=json.dumps(order) )
使用场景：不同优先级的订单（VIP用户、普通用户、批量订单）分开处理。
三、Topic Exchange：模糊匹配
Topic Exchange根据通配符匹配Routing Key。
python
class TopicExchangeExample: def init(self): self.channel.exchange_declare( exchange='order_topic', exchange_type='topic' ) def setup_queues(self): # 日本订单队列 self.channel.queue_declare('order_jp') self.channel.queue_bind( exchange='order_topic', queue='order_jp', routing_key='order.jp.' ) # 美国订单队列 self.channel.queue_declare('order_us') self.channel.queue_bind( exchange='order_topic', queue='order_us', routing_key='order.us.*' ) # 所有订单的日志队列 self.channel.queue_declare('order_log') self.channel.queue_bind( exchange='order_topic', queue='order_log', routing_key='order.#' # #匹配任意多个词 )
使用场景：按地域（国家、地区）路由订单，同时有一个全局日志队列。
四、Fanout Exchange：广播
Fanout Exchange把消息广播给所有绑定的队列。
python
class FanoutExchangeExample: def init(self): self.channel.exchange_declare( exchange='order_fanout', exchange_type='fanout' ) def setup_queues(self): # 所有消费者都收到同样的消息 self.channel.queue_declare('order_storage') # 存储服务 self.channel.queue_declare('order_notify') # 通知服务 self.channel.queue_declare('order_analytics') # 分析服务 self.channel.queue_bind('order_storage', 'order_fanout') self.channel.queue_bind('order_notify', 'order_fanout') self.channel.queue_bind('order_analytics', 'order_fanout') def publish_order(self, order): # 所有绑定的队列都会收到 self.channel.basic_publish( exchange='order_fanout', routing_key='', # Fanout忽略routing_key body=json.dumps(order) )
使用场景：订单状态变更需要通知多个服务（存储、通知、分析、日志）。
五、Headers Exchange：复杂条件匹配
Headers Exchange根据消息头中的键值对匹配，支持更复杂的条件。
python
class HeadersExchangeExample: def init(self): self.channel.exchange_declare( exchange='order_headers', exchange_type='headers' ) def setup_queues(self): # 大额订单队列 self.channel.queue_declare('order_large') self.channel.queue_bind( exchange='order_headers', queue='order_large', arguments={ 'x-match': 'all', # 所有条件都满足 'amount': 'large', 'status': 'paid' } ) # 紧急订单队列 self.channel.queue_declare('order_urgent') self.channel.queue_bind( exchange='order_headers', queue='order_urgent', arguments={ 'x-match': 'any', # 任一条件满足 'urgent': 'true', 'priority': 'high' } )
六、总结
四种交换器的选择原则：Direct用于精准路由、Topic用于模糊匹配、Fanout用于广播、Headers用于复杂条件。在跨境电商订单系统中，Direct用得最多（按订单类型路由），Fanout用于状态广播。
