---
title: "02-getProxy-setProxyRecord"
created: 2026-04-01
tags:
  - 项目
aliases:
  - getProxy-setProxyRecord
---
# getProxy/setProxyRecord

## **一、整体架构**

![[image-e11f53cb.png]]

## 二、getProxy **—— 获取代理**

```python
# -*- 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 请求流程**

![[image-de3f13ce.png]]

### **2.2 代码逐层解析**

#### **入口：Tornado 异步处理**

```python
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 是单线程异步框架，数据库查询是阻塞操作。放到线程池里执行，避免阻塞事件循环。*

#### **初始化参数**

```python
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()    # ⭐ 核心数据层实例
```

#### **参数校验**

```python
def __checkParam(self):
    if not self.channel:
        raise ParseError(...)    # channel 必填
```

#### **获取代理（核心）**

```python
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()` **—— 获取代理**

![[diagram-1775099044892-ba299498.png]]

#### **核心 SQL 拆解**

```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                     -- 猎聘特有：已开放
"""
```

#### **返回值的两种边界情况**

```text
# 情况1：完全没有代理（连池子都是空的）
if not allProxyInfos:
    return { 'proxies': [], 'proxy': '', 'limitCount': 0, 'realProxy': '' }

# 情况2：有代理但全被占用/封禁（返回所有代理列表，但不返回可用的）
if not proxyInfos:
    return { 'proxies': allProxys, 'proxy': '', 'limitCount': 0, 'realProxy': '' }
#                     ↑ 给客户端参考     ↑ 空的，表示没有可用代理
```

> 客户端收到 proxy='' 后就知道当前无可用代理，需要等待。

#### `__lockedProxy()` **—— 代理锁（防并发抢占）**

```python
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'
    """
```

**时序图：**

![[image-f3bfb19b.png]]

## **三、setProxyRecord —— 标记代理状态**

```python
# -*- 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 请求流程**

![[image-4508566d.png]]

### **3.2 代码解析**

#### **初始化参数**

```python
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
    }
```

#### **参数校验**

```python
def __checkParam(self):
    if not self.channel or not self.proxy:    # channel 和 proxy 都必填
        raise ParseError(...)
```

#### **标记代理状态**

```python
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()` **—— 标记代理使用记录**

![[image-713bc48a.png]]

#### **limitedType 的两种含义**

```text
limitedType = 1 (LIMITED_TYPE_PLATFORM)
────────────────────────────────────────
含义：被目标平台（如猎聘）识别并封禁了
动作：ProxyPool.liepin += 1（连续被封天数+1）
后果：累计到 5 → 该代理在该渠道永久排除 + 发邮件

limitedType = 2 (LIMITED_TYPE_SPIDER_COUNT)  
────────────────────────────────────────
含义：正常抓完了今天的配额
动作：ProxyPool.liepin = 0（重置！因为今天能用 = 没被封）
后果：明天继续可用
```

**关键洞察：** `limitedType=2` 时重置封禁计数，这意味着：

```text
Day1: 被封 → liepin=1
Day2: 被封 → liepin=2  
Day3: 正常用完 → liepin=0  ← 重置了！说明解封了
Day4: 被封 → liepin=1      ← 从头计数
```

> 只有连续 5 天被封才会触发告警，中间有一天正常就清零。

#### **心跳续命机制**

```sql
# 每次 setProxyRecord 都会延长锁的过期时间
sql = f'''
    UPDATE ProxyLock 
    SET ExpireAt = ExpireAt + INTERVAL 20 SECOND 
    WHERE Channel = "{channel}" 
      AND UseType = {useType} 
      AND Proxy = "{proxy}"
'''
```

```text
时间线：
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 释放 → 其他爬虫可以用了
```

> ***设计巧妙***：不需要显式释放锁，靠心跳续命 + 自动过期实现。爬虫崩溃了锁也会自动释放。

#### **邮件告警**

```python
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()` **—— 链接分片**

```python
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
```

**用途：多进程分片抓取**

```text
假设 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条)
```

*简单的取模分片，无需分布式协调，每个进程独立负责自己的分片。*

![[image-00d11fe6.png]]

---

**项目分区导航**：[[01-Spider.py基类深度解析|Spider.py基类深度解析]] ⬅️ | 02-getProxy-setProxyRecord | ➡️ [[03-spider.py|spider.py]]