update

Author: HackerYunen

api/api.py

@@ -1,9 +1,13 @@
 # encoding: utf-8
 
 import os
+import logging
 from flask import Flask
 from flask import jsonify, request, redirect, send_from_directory
 
+log = logging.getLogger('werkzeug')
+log.disabled = True
+
 try:
     from db import conn
 except:
@@ -123,9 +127,11 @@ def after_request(resp):
     return resp
 app.after_request(after_request)
 
-def main():
+def main(proc_lock):
+    if proc_lock is not None:
+        conn.set_proc_lock(proc_lock)
     # 因为默认sqlite3中，同一个数据库连接不能在多线程环境下使用，所以这里需要禁用flask的多线程
     app.run(host='0.0.0.0', port=5000, threaded=False)
 
 if __name__ == '__main__':
-    main()
+    main(None)

config.py

@@ -17,16 +17,16 @@
 PROC_VALIDATOR_SLEEP = 5
 
 # 验证器的配置参数
-VALIDATE_THREAD_NUM = 80 # 验证线程数量
+VALIDATE_THREAD_NUM = 200 # 验证线程数量
 # 验证器的逻辑是：
 # 使用代理访问 VALIDATE_URL 网站，超时时间设置为 VALIDATE_TIMEOUT
 # 如果没有超时：
 # 1、若选择的验证方式为GET：  返回的网页中包含 VALIDATE_KEYWORD 文字，那么就认为本次验证成功
 # 2、若选择的验证方式为HEAD： 返回的响应头中，对于的 VALIDATE_HEADER 响应字段内容包含 VALIDATE_KEYWORD 内容，那么就认为本次验证成功
 # 上述过程最多进行 VALIDATE_MAX_FAILS 次，只要有一次成功，就认为代理可用
-VALIDATE_URL = 'https://www.baidu.com'
-VALIDATE_METHOD = 'GET' # 验证方式，可选：GET、HEAD
-VALIDATE_HEADER = 'Server' # 仅用于HEAD验证方式，百度响应头Server字段KEYWORD可填：bfe
-VALIDATE_KEYWORD = '百度一下，你就知道'
+VALIDATE_URL = 'https://qq.com'
+VALIDATE_METHOD = 'HEAD' # 验证方式，可选：GET、HEAD
+VALIDATE_HEADER = 'location' # 仅用于HEAD验证方式，百度响应头Server字段KEYWORD可填：bfe
+VALIDATE_KEYWORD = 'www.qq.com'
 VALIDATE_TIMEOUT = 5 # 超时时间，单位s
 VALIDATE_MAX_FAILS = 3

db/Proxy.py

@@ -99,18 +99,18 @@ def validate(self, success, latency):
             self.validated = True
             self.validate_date = datetime.datetime.now()
             self.validate_failed_cnt = 0
-            self.to_validate_date = datetime.datetime.now() + datetime.timedelta(minutes=5) # 5分钟之后继续验证
+            self.to_validate_date = datetime.datetime.now() + datetime.timedelta(minutes=10)  # 10分钟之后继续验证
             return False
         else:
             self.validated = False
             self.validate_date = datetime.datetime.now()
             self.validate_failed_cnt = self.validate_failed_cnt + 1
 
             # 验证失败的次数越多，距离下次验证的时间越长
-            delay_minutes = self.validate_failed_cnt * 5
+            delay_minutes = self.validate_failed_cnt * 10
             self.to_validate_date = datetime.datetime.now() + datetime.timedelta(minutes=delay_minutes)
 
-            if self.validate_failed_cnt >= 3:
+            if self.validate_failed_cnt >= 6:
                 return True
             else:
                 return False

db/conn.py

@@ -9,9 +9,21 @@
 from .Fetcher import Fetcher
 import sqlite3
 import datetime
-import time
+import threading
 
 conn = sqlite3.connect(DATABASE_PATH, detect_types=sqlite3.PARSE_DECLTYPES | sqlite3.PARSE_COLNAMES)
+# 线程锁
+conn_lock = threading.Lock()
+# 进程锁
+proc_lock = None
+
+def set_proc_lock(proc_lock_sub):
+    """
+    设置进程锁
+    proc_lock_sub : main中的进程锁
+    """
+    global proc_lock
+    proc_lock = proc_lock_sub
 
 def pushNewFetch(fetcher_name, protocol, ip, port):
     """
@@ -21,13 +33,13 @@ def pushNewFetch(fetcher_name, protocol, ip, port):
     ip : 代理IP地址
     port : 代理端口
     """
-    time.sleep(0.1) # 为了解决并发读写饿死的问题
-
     p = Proxy()
     p.fetcher_name = fetcher_name
     p.protocol = protocol
     p.ip = ip
     p.port = port
+    conn_lock.acquire()
+    proc_lock.acquire()
 
     c = conn.cursor()
     c.execute('BEGIN EXCLUSIVE TRANSACTION;')
@@ -43,6 +55,8 @@ def pushNewFetch(fetcher_name, protocol, ip, port):
         c.execute('INSERT INTO proxies VALUES (?,?,?,?,?,?,?,?,?)', p.params())
     c.close()
     conn.commit()
+    conn_lock.release()
+    proc_lock.release()
 
 def getToValidate(max_count=1):
     """
@@ -51,6 +65,8 @@ def getToValidate(max_count=1):
     max_count : 返回数量限制
     返回 : list[Proxy]
     """
+    conn_lock.acquire()
+    proc_lock.acquire()
     c = conn.cursor()
     c.execute('BEGIN EXCLUSIVE TRANSACTION;')
     c.execute('SELECT * FROM proxies WHERE to_validate_date<=? AND validated=? ORDER BY to_validate_date LIMIT ?', (
@@ -67,6 +83,8 @@ def getToValidate(max_count=1):
     proxies = proxies + [Proxy.decode(row) for row in c]
     c.close()
     conn.commit()
+    conn_lock.release()
+    proc_lock.release()
     return proxies
 
 def pushValidateResult(proxy, success, latency):
@@ -76,10 +94,10 @@ def pushValidateResult(proxy, success, latency):
     success : True/False，验证是否成功
     latency : 本次验证所用的时间(单位毫秒)
     """
-    time.sleep(0.01) # 为了解决并发读写饿死的问题
-
     p = proxy
     should_remove = p.validate(success, latency)
+    conn_lock.acquire()
+    proc_lock.acquire()
     if should_remove:
         conn.execute('DELETE FROM proxies WHERE protocol=? AND ip=? AND port=?', (p.protocol, p.ip, p.port))
     else:
@@ -92,19 +110,25 @@ def pushValidateResult(proxy, success, latency):
             p.protocol, p.ip, p.port
         ))
     conn.commit()
+    conn_lock.release()
+    proc_lock.release()
 
 def getValidatedRandom(max_count):
     """
     从通过了验证的代理中，随机选择max_count个代理返回
     max_count<=0表示不做数量限制
     返回 : list[Proxy]
     """
+    conn_lock.acquire()
+    proc_lock.acquire()
     if max_count > 0:
         r = conn.execute('SELECT * FROM proxies WHERE validated=? ORDER BY RANDOM() LIMIT ?', (True, max_count))
     else:
         r = conn.execute('SELECT * FROM proxies WHERE validated=? ORDER BY RANDOM()', (True,))
     proxies = [Proxy.decode(row) for row in r]
     r.close()
+    conn_lock.release()
+    proc_lock.release()
     return proxies
 
 def pushFetcherResult(name, proxies_cnt):
@@ -113,8 +137,8 @@ def pushFetcherResult(name, proxies_cnt):
     name : 爬取器的名称
     proxies_cnt : 本次爬取到的代理数量
     """
-    time.sleep(0.1) # 为了解决并发读写饿死的问题
-
+    conn_lock.acquire()
+    proc_lock.acquire()
     c = conn.cursor()
     c.execute('BEGIN EXCLUSIVE TRANSACTION;')
     c.execute('SELECT * FROM fetchers WHERE name=?', (name,))
@@ -131,13 +155,17 @@ def pushFetcherResult(name, proxies_cnt):
         ))
     c.close()
     conn.commit()
+    conn_lock.release()
+    proc_lock.release()
 
 def pushFetcherEnable(name, enable):
     """
     设置是否起用对应爬取器，被禁用的爬取器将不会被运行
     name : 爬取器的名称
     enable : True/False, 是否启用
     """
+    conn_lock.acquire()
+    proc_lock.acquire()
     c = conn.cursor()
     c.execute('BEGIN EXCLUSIVE TRANSACTION;')
     c.execute('SELECT * FROM fetchers WHERE name=?', (name,))
@@ -152,25 +180,35 @@ def pushFetcherEnable(name, enable):
         ))
     c.close()
     conn.commit()
+    conn_lock.release()
+    proc_lock.release()
 
 def getAllFetchers():
     """
     获取所有的爬取器以及状态
     返回 : list[Fetcher]
     """
+    conn_lock.acquire()
+    proc_lock.acquire()
     r = conn.execute('SELECT * FROM fetchers')
     fetchers = [Fetcher.decode(row) for row in r]
     r.close()
+    conn_lock.release()
+    proc_lock.release()
     return fetchers
 
 def getFetcher(name):
     """
     获取指定爬取器以及状态
     返回 : Fetcher
     """
+    conn_lock.acquire()
+    proc_lock.acquire()
     r = conn.execute('SELECT * FROM fetchers WHERE name=?', (name,))
     row = r.fetchone()
     r.close()
+    conn_lock.release()
+    proc_lock.release()
     if row is None:
         return None
     else:
@@ -182,16 +220,22 @@ def getProxyCount(fetcher_name):
     fetcher_name : 爬取器名称
     返回 : int
     """
+    conn_lock.acquire()
+    proc_lock.acquire()
     r = conn.execute('SELECT count(*) FROM proxies WHERE fetcher_name=?', (fetcher_name,))
     cnt = r.fetchone()[0]
     r.close()
+    conn_lock.release()
+    proc_lock.release()
     return cnt
 
 def getProxiesStatus():
     """
     获取代理状态，包括`全部代理数量`，`当前可用代理数量`，`等待验证代理数量`
     返回 : dict
     """
+    conn_lock.acquire()
+    proc_lock.acquire()
     r = conn.execute('SELECT count(*) FROM proxies')
     sum_proxies_cnt = r.fetchone()[0]
     r.close()
@@ -203,7 +247,8 @@ def getProxiesStatus():
     r = conn.execute('SELECT count(*) FROM proxies WHERE to_validate_date<=?', (datetime.datetime.now(),))
     pending_proxies_cnt = r.fetchone()[0]
     r.close()
-
+    conn_lock.release()
+    proc_lock.release()
     return dict(
         sum_proxies_cnt=sum_proxies_cnt,
         validated_proxies_cnt=validated_proxies_cnt,
@@ -214,8 +259,12 @@ def pushClearFetchersStatus():
     """
     清空爬取器的统计信息，包括sum_proxies_cnt,last_proxies_cnt,last_fetch_date
     """
+    conn_lock.acquire()
+    proc_lock.acquire()
     c = conn.cursor()
     c.execute('BEGIN EXCLUSIVE TRANSACTION;')
     c.execute('UPDATE fetchers SET sum_proxies_cnt=?, last_proxies_cnt=?, last_fetch_date=?', (0, 0, None))
     c.close()
     conn.commit()
+    conn_lock.release()
+    proc_lock.release()

main.py

@@ -6,6 +6,10 @@
 import time
 from proc import run_fetcher, run_validator
 from api import api
+import multiprocessing
+
+# 进程锁
+proc_lock = multiprocessing.Lock()
 
 class Item:
     def __init__(self, target, name):
@@ -23,7 +27,7 @@ def main():
     while True:
         for p in processes:
             if p.process is None:
-                p.process = Process(target=p.target, name=p.name, daemon=False)
+                p.process = Process(target=p.target, name=p.name, daemon=False, args=(proc_lock, ))
                 p.process.start()
                 print(f'启动{p.name}进程，pid={p.process.pid}')
                 p.start_time = time.time()
@@ -56,7 +60,7 @@ def citest():
         p.process.start()
         print(f'running {p.name}, pid={p.process.pid}')
         p.start_time = time.time()
-    
+
     time.sleep(10)
 
     for p in processes:

proc/run_fetcher.py

@@ -11,10 +11,12 @@
 from db import conn
 from fetchers import fetchers
 from config import PROC_FETCHER_SLEEP
+from func_timeout import func_set_timeout
+from func_timeout.exceptions import FunctionTimedOut
 
 logging.basicConfig(stream=sys.stdout, format="%(asctime)s-%(levelname)s:%(name)s:%(message)s", level='INFO')
 
-def main():
+def main(proc_lock):
     """
     定时运行爬取器
     主要逻辑：
@@ -27,6 +29,8 @@ def main():
         睡眠一段时间
     """
     logger = logging.getLogger('fetcher')
+    conn.set_proc_lock(proc_lock)
+
     while True:
         logger.info('开始运行一轮爬取器')
         status = conn.getProxiesStatus()
@@ -35,19 +39,27 @@ def main():
             time.sleep(PROC_FETCHER_SLEEP)
             continue
 
+        @func_set_timeout(30)
+        def fetch_worker(fetcher):
+            f = fetcher()
+            proxies = f.fetch()
+            return proxies
+
         def run_thread(name, fetcher, que):
             """
             name: 爬取器名称
             fetcher: 爬取器class
             que: 队列，用于返回数据
             """
             try:
-                f = fetcher()
-                proxies = f.fetch()
+                proxies = fetch_worker(fetcher)
                 que.put((name, proxies))
             except Exception as e:
                 logger.error(f'运行爬取器{name}出错：' + str(e))
                 que.put((name, []))
+            except FunctionTimedOut:
+                pass
+
         threads = []
         que = Queue()
         for item in fetchers:
@@ -61,8 +73,7 @@ def run_thread(name, fetcher, que):
             threads.append(threading.Thread(target=run_thread, args=(item.name, item.fetcher, que)))
         [t.start() for t in threads]
         [t.join() for t in threads]
-        for _ in range(len(threads)):
-            assert not que.empty()
+        while not que.empty():
             fetcher_name, proxies = que.get()
             for proxy in proxies:
                 conn.pushNewFetch(fetcher_name, proxy[0], proxy[1], proxy[2])