Merge remote-tracking branch 'upstream/master'

上游更改DB工厂类接口

DB/SsdbClient.py

@@ -10,6 +10,7 @@
    Change Activity:
                    2016/12/2:
                    2017/09/22: PY3中 redis-py返回的数据是bytes型
+                   2017/09/27: 修改pop()方法 返回{proxy:value}字典
 -------------------------------------------------
 """
 __author__ = 'JHao'
@@ -77,11 +78,17 @@ class SsdbClient(object):
         self.__conn.hincrby(self.name, key, value)
 
     def pop(self):
+        """
+        弹出一个代理
+        :return: dict {proxy: value}
+        """
         proxies = self.__conn.hkeys(self.name)
         if proxies:
             proxy = random.choice(proxies)
+            value = self.__conn.hget(self.name, proxy)
             self.delete(proxy)
-            return proxy
+            return {'proxy': proxy.decode('utf-8') if EnvUtil.PY3 else proxy,
+                    'value': value.decode('utf-8') if EnvUtil.PY3 and value else value}
         return None
 
     def exists(self, key):

Schedule/ProxyCheck.py

@@ -0,0 +1,62 @@
+# -*- coding: utf-8 -*-
+"""
+-------------------------------------------------
+   File Name：     ProxyCheck
+   Description :   多线程验证useful_proxy
+   Author :        J_hao
+   date：          2017/9/26
+-------------------------------------------------
+   Change Activity:
+                   2017/9/26: 多线程验证useful_proxy
+-------------------------------------------------
+"""
+__author__ = 'J_hao'
+
+import sys
+from time import sleep
+from threading import Thread
+
+sys.path.append('../')
+
+from Util.utilFunction import validUsefulProxy
+from Manager.ProxyManager import ProxyManager
+from Util.LogHandler import LogHandler
+
+
+class ProxyCheck(ProxyManager, Thread):
+
+    def __init__(self):
+        ProxyManager.__init__(self)
+        Thread.__init__(self)
+        self.log = LogHandler('proxy_check')
+
+    def run(self):
+        self.db.changeTable(self.useful_proxy_queue)
+        while True:
+            proxy_item = self.db.pop()
+            while proxy_item:
+                proxy = proxy_item.get('proxy')
+                counter = proxy_item.get('value')
+                if validUsefulProxy(proxy):
+                    # 验证通过计数器加1, 计数在-5到1之间
+                    if counter and int(counter) < 1:
+                        self.db.put(proxy, num=int(counter) + 1)
+                    else:
+                        self.db.put(proxy)
+                    self.log.info('ProxyCheck: {} validation pass'.format(proxy))
+                else:
+                    self.log.info('ProxyCheck: {} validation fail'.format(proxy))
+                    # 验证失败，计数器减1
+                    if counter and int(counter) < -5:
+                        self.log.info('ProxyCheck: {} fail too many, delete!'.format(proxy))
+                        self.db.delete(proxy)
+                    else:
+                        self.db.put(proxy, num=int(counter) - 1)
+
+                proxy_item = self.db.pop()
+            sleep(60 * 5)
+
+
+if __name__ == '__main__':
+    p = ProxyCheck()
+    p.run()

Schedule/ProxyRefreshSchedule.py

@@ -46,11 +46,12 @@ class ProxyRefreshSchedule(ProxyManager):
         :return:
         """
         self.db.changeTable(self.raw_proxy_queue)
-        raw_proxy = self.db.pop()
+        raw_proxy_item = self.db.pop()
         self.log.info('ProxyRefreshSchedule: %s start validProxy' % time.ctime())
         # 计算剩余代理，用来减少重复计算
         remaining_proxies = self.getAll()
-        while raw_proxy:
+        while raw_proxy_item:
+            raw_proxy = raw_proxy_item.get('proxy')
             if isinstance(raw_proxy, bytes):
                 # 兼容Py3
                 raw_proxy = raw_proxy.decode('utf8')
@@ -62,7 +63,7 @@ class ProxyRefreshSchedule(ProxyManager):
             else:
                 self.log.info('ProxyRefreshSchedule: %s validation fail' % raw_proxy)
             self.db.changeTable(self.raw_proxy_queue)
-            raw_proxy = self.db.pop()
+            raw_proxy_item = self.db.pop()
             remaining_proxies = self.getAll()
         self.log.info('ProxyRefreshSchedule: %s validProxy complete' % time.ctime())
 

Schedule/ProxyValidSchedule.py

@@ -13,49 +13,31 @@
 __author__ = 'JHao'
 
 import sys
-from time import sleep
 
 sys.path.append('../')
 
-from Util.utilFunction import validUsefulProxy
-from Manager.ProxyManager import ProxyManager
-from Util.LogHandler import LogHandler
+from Schedule.ProxyCheck import ProxyCheck
 
 
-class ProxyValidSchedule(ProxyManager):
+class ProxyValidSchedule(object):
     def __init__(self):
-        ProxyManager.__init__(self)
-        self.log = LogHandler('valid_schedule')
+        pass
 
-    def __validProxy(self):
+    def __validProxy(self, threads=5):
         """
-        验证代理
+        验证useful_proxy代理
+        :param threads: 线程数
         :return:
         """
-        while True:
-            self.db.changeTable(self.useful_proxy_queue)
-            for each_proxy in self.db.getAll():
-                if isinstance(each_proxy, bytes):
-                    # 兼容PY3
-                    each_proxy = each_proxy.decode('utf-8')
+        thread_list = list()
+        for index in range(threads):
+            thread_list.append(ProxyCheck())
 
-                value = self.db.get(each_proxy)
-                if validUsefulProxy(each_proxy):
-                    # 成功计数器加1
-                    if value and int(value) < 1:
-                        self.db.update(each_proxy, 1)
-                    self.log.info('ProxyValidSchedule: {} validation pass'.format(each_proxy))
-                else:
-                    # 失败计数器减一
-                    if value and int(value) < -5:
-                        # 计数器小于-5删除该代理
-                        self.db.delete(each_proxy)
-                    else:
-                        self.db.update(each_proxy, -1)
-                    self.log.info('ProxyValidSchedule: {} validation fail'.format(each_proxy))
+        for thread in thread_list:
+            thread.start()
 
-            self.log.info('ProxyValidSchedule running normal')
-            sleep(60 * 1)
+        for thread in thread_list:
+            thread.join()
 
     def main(self):
         self.__validProxy()

doc/release_notes.md

@@ -1,5 +1,8 @@
 ## Release Notes
 
+* newest
+　　1.使用多线程验证useful_pool
+
 * 1.10
 　　1. 第一版；
 　　2. 支持PY2/PY3;

log/test.log

@@ -1 +0,0 @@
-2017-04-24 15:34:08,434 LogHandler.py[line:85] ERROR this is a test msg