getProxy/setProxyRecord
一、整体架构
二、getProxy —— 获取代理
# -*- coding: utf-8 -*-
"""
@Desc : 获取可用代理
@Date : 2024/7/18 11:31
@Author : gaofenfei
"""
import tornado.web
from tornado.concurrent import run_on_executor
from concurrent.futures import ThreadPoolExecutor
from __core import ToolHandler
from __core import LoggingHandler
from api import ParseError
from api import ParseConst
from __thirdapi import ApiError
from api.proxypool import Data
import os
class getProxy(tornado.web.RequestHandler):
executor = ThreadPoolExecutor(5)
@tornado.gen.coroutine
def post(self):
yield self.run()
self.finish()
@run_on_executor
def run(self):
# 初始化
self.__init()
try:
# 校验参数
self.__checkParam()
self.__getProxy()
except ParseError as e:
self.logger.error({'code': e.getCode(), 'msg': e.getMsg(), 'data': e.getData()}, exc_info=True)
ToolHandler.apiToError(self, e.getCode(), e.getMsg(), e.getData())
except Exception as e:
self.logger.error({'error': e, 'env':self.env, 'channel':self.channel,'useType': self.useType},exc_info=True)
ToolHandler.apiToError(self, 500, e)
finally:
if self.data.mysqlSearchJob:
self.data.mysqlSearchJob.close()
def __getProxy(self):
'''获取代理'''
rs = self.data.getProxy(self.channel,self.env,self.useType,self.nw)
rs = {'code': 200, 'msg': 'ok', 'data': rs}
self.write(rs)
def __checkParam(self):
'''校验参数'''
if not self.channel:
errorData = {}
errorData['channel'] = 'Yes' if self.channel else 'No'
raise ParseError(ParseError.PARAMS.get('msg'), ParseError.PARAMS.get('code'),errorData)
def __init(self):
'''初始化'''
self.channel = self.get_argument('channel', '') # 渠道
self.env = self.get_argument('env', ParseConst.ENV_TYPE_LOCAL) # 平台,默认本地
self.useType = self.get_argument('useType',ParseConst.USE_TYPE_JOB) # 用途,默认职位
self.nw = int(self.get_argument('nw',0)) # 是否返回内网IP，默认是0
# 初始化日志文件
self.loggerName = os.path.basename(__file__).split('.')[0]
self.logger = LoggingHandler(self.loggerName, 'ERROR', True, self.loggerName)
self.data = Data()
if __name__ == '__main__':
exit(1)
2.1 请求流程
2.2 代码逐层解析
入口：Tornado 异步处理
class getProxy(tornado.web.RequestHandler):
executor = ThreadPoolExecutor(5) # 线程池，最多5个并发
@tornado.gen.coroutine
def post(self):
yield self.run() # 异步等待
self.finish() # 结束响应
@run_on_executor # 放到线程池执行，不阻塞主线程
def run(self):
self.__init()
try:
self.__checkParam()
self.__getProxy()
except ParseError as e: # 业务异常（参数错误等）
...
except Exception as e: # 系统异常
...
finally:
if self.data.mysqlSearchJob:
self.data.mysqlSearchJob.close() # ⭐ 确保关闭数据库连接
为什么用
run_on_executor？ Tornado 是单线程异步框架，数据库查询是阻塞操作。放到线程池里执行，避免阻塞事件循环。
初始化参数
def __init(self):
self.channel = self.get_argument('channel', '') # 渠道（liepin/boss等）
self.env = self.get_argument('env', ParseConst.ENV_TYPE_LOCAL) # 环境（本地/线上）
self.useType = self.get_argument('useType', ParseConst.USE_TYPE_JOB) # 用途（抓职位/抓公司）
self.nw = int(self.get_argument('nw', 0)) # 是否返回内网IP
self.data = Data() # ⭐ 核心数据层实例
参数校验
def __checkParam(self):
if not self.channel:
raise ParseError(...) # channel 必填
获取代理（核心）
def __getProxy(self):
# 真正的逻辑在 Data 层
rs = self.data.getProxy(self.channel, self.env, self.useType, self.nw)
rs = {'code': 200, 'msg': 'ok', 'data': rs}
self.write(rs)
2.3 Data().getProxy() 做了什么
getProxy() —— 获取代理
核心 SQL 拆解
# ── 被锁的代理（其他爬虫正在使用）──
lockedSql = """
SELECT Proxy FROM ProxyLock
WHERE Channel = 'liepin'
AND UseType = 1
AND ExpireAt > NOW() -- 锁还没过期
"""
# ── 被封或达到限额的代理 ──
limitedSql = """
SELECT Proxy FROM ProxyPoolRecord
WHERE Channel = 'liepin'
AND UseType = 1
AND (
LimitedType = 1 -- 被平台封了
OR (LimitedType = 2 AND SpiderCount >= 100) -- 达到每日限额
)
AND LimitedDate = CURDATE() -- 只看今天
"""
# ── 合并排除列表 ──
passProxy = f"{lockedSql} UNION ALL {limitedSql}"
# ── 最终查询可用代理 ──
sql = """
SELECT * FROM ProxyPool
WHERE Status = 0 -- 状态可用
AND UseEnv IN (1,3) -- 环境匹配
AND liepin < 5 -- 该渠道连续被封 < 5次
AND UseType = 1 -- 用途匹配
AND Proxy NOT IN (排除列表) -- 不在锁定/封禁列表中
AND Open = 1 -- 猎聘特有：已开放
"""
返回值的两种边界情况
# 情况1：完全没有代理（连池子都是空的）
if not allProxyInfos:
return { 'proxies': [], 'proxy': '', 'limitCount': 0, 'realProxy': '' }
# 情况2：有代理但全被占用/封禁（返回所有代理列表，但不返回可用的）
if not proxyInfos:
return { 'proxies': allProxys, 'proxy': '', 'limitCount': 0, 'realProxy': '' }
# ↑ 给客户端参考 ↑ 空的，表示没有可用代理
客户端收到 proxy='' 后就知道当前无可用代理，需要等待。
__lockedProxy() —— 代理锁（防并发抢占）
def __lockedProxy(self, channel, useType, proxy):
lockedAt = datetime.now()
expireAt = datetime.now() + timedelta(minutes=10) # 10分钟后过期
sql = """
INSERT INTO ProxyLock (Channel, UseType, Proxy, LockedAt, ExpireAt)
VALUES ('liepin', 1, 'ip:port', '2024-07-03 19:31:00', '2024-07-03 19:41:00')
ON DUPLICATE KEY UPDATE
LockedAt = '2024-07-03 19:31:00',
ExpireAt = '2024-07-03 19:41:00'
"""
时序图：
三、setProxyRecord —— 标记代理状态
# -*- coding: utf-8 -*-
"""
@Desc : 设置代理不可用
@Date : 2024/7/18 11:31
@Author : gaofenfei
"""
import tornado.web
from tornado.concurrent import run_on_executor
from concurrent.futures import ThreadPoolExecutor
from __core import ToolHandler
from __core import LoggingHandler
from api import ParseError
from api import ParseConst
from __thirdapi import ApiError
from api.proxypool import Data
import os
class setProxyRecord(tornado.web.RequestHandler):
executor = ThreadPoolExecutor(5)
@tornado.gen.coroutine
def post(self):
yield self.run()
self.finish()
@run_on_executor
def run(self):
# 初始化
self.__init()
try:
# 校验参数
self.__checkParam()
self.__setProxyRecord()
except ParseError as e:
self.logger.error({'code': e.getCode(), 'msg': e.getMsg(), 'data': e.getData()}, exc_info=True)
ToolHandler.apiToError(self, e.getCode(), e.getMsg(), e.getData())
except Exception as e:
errorData = {'error':e}
errorData.update(self.loggerParam)
self.logger.error(errorData,exc_info=True)
ToolHandler.apiToError(self, 500, e)
finally:
if self.data.mysqlSearchJob:
self.data.mysqlSearchJob.close()
def __setProxyRecord(self):
'''获取代理'''
rs = self.data.setProxyRecord(self.channel,self.proxy,self.useType,self.limitedType,self.spiderCount)
rs = {'code': 200, 'msg': 'ok', 'data': 'success'}
self.write(rs)
def __checkParam(self):
'''校验参数'''
if not self.channel or not self.proxy:
errorData = {}
errorData['channel'] = 'Yes' if self.channel else 'No'
errorData['proxy'] = 'Yes' if self.proxy else 'No'
raise ParseError(ParseError.PARAMS.get('msg'), ParseError.PARAMS.get('code'),errorData)
def __init(self):
'''初始化'''
self.channel = self.get_argument('channel', '') # 渠道
self.proxy = self.get_argument('proxy', '') # 代理
self.useType = self.get_argument('useType',ParseConst.USE_TYPE_JOB) # 用途,默认职位
self.limitedType = self.get_argument('limitedType',ParseConst.LIMITED_TYPE_PLATFORM) # 限制类型，默认平台
self.spiderCount = int(self.get_argument('spiderCount',0))
# 初始化日志文件
self.loggerName = os.path.basename(__file__).split('.')[0]
self.logger = LoggingHandler(self.loggerName, 'ERROR', True, self.loggerName)
self.data = Data()
self.loggerParam = {
'channel': self.channel,
'proxy': self.proxy,
'useType': self.useType,
'limitedType': self.limitedType,
'spiderCount': self.spiderCount
}
if __name__ == '__main__':
exit(1)
3.1 请求流程
3.2 代码解析
初始化参数
def __init(self):
self.channel = self.get_argument('channel', '')
self.proxy = self.get_argument('proxy', '')
self.useType = self.get_argument('useType', ParseConst.USE_TYPE_JOB)
self.limitedType = self.get_argument('limitedType', ParseConst.LIMITED_TYPE_PLATFORM)
self.spiderCount = int(self.get_argument('spiderCount', 0))
# ⭐ 日志参数（出错时记录完整上下文）
self.loggerParam = {
'channel': self.channel,
'proxy': self.proxy,
'useType': self.useType,
'limitedType': self.limitedType,
'spiderCount': self.spiderCount
}
参数校验
def __checkParam(self):
if not self.channel or not self.proxy: # channel 和 proxy 都必填
raise ParseError(...)
标记代理状态
def __setProxyRecord(self):
rs = self.data.setProxyRecord(
self.channel, # 渠道
self.proxy, # 代理地址
self.useType, # 用途
self.limitedType, # 受限类型
self.spiderCount # 已抓取数量
)
rs = {'code': 200, 'msg': 'ok', 'data': 'success'}
self.write(rs)
3.3 Data().setProxyRecord() 做了什么
setProxyRecord() —— 标记代理使用记录
limitedType 的两种含义
limitedType = 1 (LIMITED_TYPE_PLATFORM)
────────────────────────────────────────
含义：被目标平台（如猎聘）识别并封禁了
动作：ProxyPool.liepin += 1（连续被封天数+1）
后果：累计到 5 → 该代理在该渠道永久排除 + 发邮件
limitedType = 2 (LIMITED_TYPE_SPIDER_COUNT)
────────────────────────────────────────
含义：正常抓完了今天的配额
动作：ProxyPool.liepin = 0（重置！因为今天能用 = 没被封）
后果：明天继续可用
关键洞察： limitedType=2 时重置封禁计数，这意味着：
Day1: 被封 → liepin=1
Day2: 被封 → liepin=2
Day3: 正常用完 → liepin=0 ← 重置了！说明解封了
Day4: 被封 → liepin=1 ← 从头计数
只有连续 5 天被封才会触发告警，中间有一天正常就清零。
心跳续命机制
# 每次 setProxyRecord 都会延长锁的过期时间
sql = f'''
UPDATE ProxyLock
SET ExpireAt = ExpireAt + INTERVAL 20 SECOND
WHERE Channel = "{channel}"
AND UseType = {useType}
AND Proxy = "{proxy}"
'''
时间线：
19:30:00 getProxy() → 锁定 ip:p1，ExpireAt = 19:40:00
19:31:00 抓了10个 → setProxyRecord() → ExpireAt = 19:40:20
19:32:00 抓了10个 → setProxyRecord() → ExpireAt = 19:40:40
19:33:00 抓了10个 → setProxyRecord() → ExpireAt = 19:41:00
...
19:35:00 爬虫崩溃了！
...
19:41:00 锁自动过期 → ip:p1 释放 → 其他爬虫可以用了
设计巧妙：不需要显式释放锁，靠心跳续命 + 自动过期实现。爬虫崩溃了锁也会自动释放。
邮件告警
if proxyInfo.get(f'{channel}') >= 5:
msg = f'{proxy} 已连续5天被 {channel} 屏蔽，请更换。'
msg += '更换sql：'
msg += f'UPDATE ProxyPool SET Proxy="新代理" ... WHERE Proxy = "{proxy}"'
mail = MailHandler()
mail.sendMail(self.mailTo.split(','), '代理超限通知', msg)
邮件里直接附带了更换 SQL，运维收到邮件复制粘贴就能操作，非常贴心。
getSpiderUrl() —— 链接分片
def getSpiderUrl(self, channel, num, index):
sql = f'SELECT * FROM SpiderLink WHERE Channel = "{channel}"'
rs = self.mysqlSearchJob.find(sql)
# ⭐ 取模分片：Id % num == index 的才返回
filtered = [row for row in rs if int(row['Id']) % num == index] if num > 0 else rs
return filtered
用途：多进程分片抓取
假设 SpiderLink 有 100 条记录（Id: 1~100）
启动 4 个爬虫进程 (num=4)：
进程0 (index=0): Id % 4 == 0 → Id=4,8,12,16... (25条)
进程1 (index=1): Id % 4 == 1 → Id=1,5,9,13... (25条)
进程2 (index=2): Id % 4 == 2 → Id=2,6,10,14... (25条)
进程3 (index=3): Id % 4 == 3 → Id=3,7,11,15... (25条)
简单的取模分片，无需分布式协调，每个进程独立负责自己的分片。
项目分区导航：Spider.py基类深度解析 ⬅️ | 02-getProxy-setProxyRecord | ➡️ spider.py
💬 评论