参考：https://www.yuanrenxue.com/crawler/news-crawler-urlpool.html

url_pool.py

# -*- coding: utf-8 -*-
# @Author  : 佛祖保佑, 永无 bug
# @Date    :
# @File    : url_pool.py
# @Software: PyCharm
# @description : XXX
import time
import redis
import pickle
import urllib.parse as urlparse
class UrlDB(object):
    """使用 redis 来存储 URL"""
    status_failure = b'0'
    status_success = b'1'
    def __init__(self, db_name):
        # self.name = db_name + '.urldb'
        # self.db = leveldb.LevelDB(self.name)
        self.name = db_name if db_name else 'redis_hashmap'
        self.db = redis.StrictRedis()
    def set_success(self, url=None):
        if isinstance(url, str):
            url = url.encode('utf8')
        try:
            self.db.hset(self.name, url, self.status_success)
            status = True
        except BaseException as be:
            status = False
        return status
    def set_failure(self, url):
        if isinstance(url, str):
            url = url.encode('utf8')
        try:
            self.db.hset(self.name, url, self.status_failure)
            status = True
        except BaseException as be:
            status = False
        return status
    def has(self, url):
        if isinstance(url, str):
            url = url.encode('utf8')
        try:
            attr = self.db.hget(self.name, url)
            return attr
        except BaseException as be:
            pass
        return False
class UrlPool(object):
    """ 使用 UrlPool 来抓取和管理URLs"""
    def __init__(self, pool_name):
        self.name = pool_name
        self.db = UrlDB(pool_name)
        self.waiting = dict()  # {host: set([urls]), } 按host分组，记录等待下载的URL
        self.pending = dict()  # {url: pended_time, } 记录已被取出（self.pop()）但还未被更新状态（正在下载）的URL
        self.failure = dict()  # {url: times,} 记录失败的URL的次数
        self.failure_threshold = 3
        self.pending_threshold = 10  # pending的最大时间，过期要重新下载
        self.waiting_count = 0  # self.waiting 字典里面的url的个数
        self.max_hosts = ['', 0]  # [host: url_count] 目前pool中url最多的host及其url数量
        self.hub_pool = dict()  # {url: last_query_time, }  存放hub url
        self.hub_refresh_span = 0
        self.load_cache()
        pass
    def __del__(self):
        self.dump_cache()
    def load_cache(self,):
        path = self.name + '.pkl'
        try:
            with open(path, 'rb') as f:
                self.waiting = pickle.load(f)
            cc = [len(v) for k, v in self.waiting.items()]
            print('saved pool loaded! urls:', sum(cc))
        except BaseException as be:
            pass
    def dump_cache(self):
        path = self.name + '.pkl'
        try:
            with open(path, 'wb') as f:
                pickle.dump(self.waiting, f)
            print('self.waiting saved!')
        except BaseException as be:
            pass
    def set_hubs(self, urls, hub_refresh_span):
        self.hub_refresh_span = hub_refresh_span
        self.hub_pool = dict()
        for url in urls:
            self.hub_pool[url] = 0
    def set_status(self, url, status_code):
        if url in self.pending:
            self.pending.pop(url)
        if status_code == 200:
            self.db.set_success(url)
            return
        if status_code == 404:
            self.db.set_failure(url)
            return
        if url in self.failure:
            self.failure[url] += 1
            if self.failure[url] > self.failure_threshold:
                self.db.set_failure(url)
                self.failure.pop(url)
            else:
                self.add(url)
        else:
            self.failure[url] = 1
            self.add(url)
    def push_to_pool(self, url=None):
        host = urlparse.urlparse(url).netloc
        if not host or '.' not in host:
            print('try to push_to_pool with bad url:', url, ', len of ur:', len(url))
            return False
        if host in self.waiting:
            if url in self.waiting[host]:
                return True
            self.waiting[host].add(url)
            if len(self.waiting[host]) > self.max_hosts[1]:
                self.max_hosts[1] = len(self.waiting[host])
                self.max_hosts[0] = host
        else:
            self.waiting[host] = set([url])
        self.waiting_count += 1
        return True
    def add(self, url=None, always=False):
        if always:
            return self.push_to_pool(url)
        pended_time = self.pending.get(url, 0)
        if time.time() - pended_time < self.pending_threshold:
            print('being downloading:', url)
            return
        if self.db.has(url):
            return
        if pended_time:
            self.pending.pop(url)
        return self.push_to_pool(url)
    def add_many(self, url_list=None, always=False):
        if isinstance(url_list, str):
            print('urls is a str !!!!', url_list)
            self.add(url_list, always)
        else:
            for url in url_list:
                self.add(url, always)
    def pop(self, count=None, hub_percent=50):
        print('\n\tmax of host:', self.max_hosts)
        # 取出的url有两种类型：hub=1, 普通=0
        url_attr_url = 0
        url_attr_hub = 1
        # 1. 首先取出hub，保证获取hub里面的最新url.
        hubs = dict()
        hub_count = count * hub_percent // 100
        for hub in self.hub_pool:
            span = time.time() - self.hub_pool[hub]
            if span < self.hub_refresh_span:
                continue
            hubs[hub] = url_attr_hub  # 1 means hub-url
            self.hub_pool[hub] = time.time()
            if len(hubs) >= hub_count:
                break
        # 2. 再取出普通url
        left_count = count - len(hubs)
        urls = dict()
        for host in self.waiting:
            if not self.waiting[host]:
                continue
            url = self.waiting[host].pop()
            urls[url] = url_attr_url
            self.pending[url] = time.time()
            if self.max_hosts[0] == host:
                self.max_hosts[1] -= 1
            if len(urls) >= left_count:
                break
        self.waiting_count -= len(urls)
        print('To pop:%s, hubs: %s, urls: %s, hosts:%s' % (count, len(hubs), len(urls), len(self.waiting)))
        urls.update(hubs)
        return urls
    def size(self,):
        return self.waiting_count
    def empty(self,):
        return self.waiting_count == 0
def test():
    pool = UrlPool('crawl_url_pool')
    urls = [
        'http://1.a.cn/xyz',
        'http://2.a.cn/xyz',
        'http://3.a.cn/xyz',
        'http://1.b.cn/xyz-1',
        'http://1.b.cn/xyz-2',
        'http://1.b.cn/xyz-3',
        'http://1.b.cn/xyz-4',
    ]
    pool.add_many(urls)
    # del pool
    # pool = UrlPool('crawl_url_pool')
    urls = pool.pop(5)
    urls = list(urls.keys())
    print('pop:', urls)
    print('pending:', pool.pending)
    pool.set_status(urls[0], 200)
    print('pending:', pool.pending)
    pool.set_status(urls[1], 404)
    print('pending:', pool.pending)
if __name__ == '__main__':
    test()

ezpymysql.py

：大规模异步新闻爬虫： 让MySQL 数据库操作更方便 - 猿人学

# file: ezpymysql.py
# Author: veelion
"""A lightweight wrapper around PyMySQL.
only for python3
"""
import time
import logging
import traceback
import pymysql
import pymysql.cursors
version = "0.7"
version_info = (0, 7, 0, 0)
class Connection(object):
    """A lightweight wrapper around PyMySQL."""
    def __init__(self, host, database, user=None, password=None,
                 port=0, max_idle_time=7 * 3600, connect_timeout=10,
                 time_zone="+0:00", charset="utf8mb4", sql_mode="TRADITIONAL"):
        self.host = host
        self.database = database
        self.max_idle_time = float(max_idle_time)
        args = dict(
            use_unicode=True, charset=charset, database=database,
            init_command=('SET time_zone = "%s"' % time_zone),
            cursorclass=pymysql.cursors.DictCursor,
            connect_timeout=connect_timeout, sql_mode=sql_mode
        )
        if user is not None:
            args["user"] = user
        if password is not None:
            args["passwd"] = password
        # We accept a path to a MySQL socket file or a host(:port) string
        if "/" in host:
            args["unix_socket"] = host
        else:
            self.socket = None
            pair = host.split(":")
            if len(pair) == 2:
                args["host"] = pair[0]
                args["port"] = int(pair[1])
            else:
                args["host"] = host
                args["port"] = 3306
        if port:
            args['port'] = port
        self._db = None
        self._db_args = args
        self._last_use_time = time.time()
        try:
            self.reconnect()
        except BaseException as be:
            logging.error("Cannot connect to MySQL on %s", self.host, exc_info=True)
    def _ensure_connected(self):
        # Mysql by default closes client connections that are idle for
        # 8 hours, but the client library does not report this fact until
        # you try to perform a query and it fails.  Protect against this
        # case by preemptively closing and reopening the connection
        # if it has been idle for too long (7 hours by default).
        if self._db is None or (time.time() - self._last_use_time > self.max_idle_time):
            self.reconnect()
        self._last_use_time = time.time()
    def _cursor(self):
        self._ensure_connected()
        return self._db.cursor()
    def __del__(self):
        self.close()
    def close(self):
        """Closes this database connection."""
        if getattr(self, "_db", None) is not None:
            self._db.close()
            self._db = None
    def reconnect(self):
        """Closes the existing database connection and re-opens it."""
        self.close()
        self._db = pymysql.connect(**self._db_args)
        self._db.autocommit(True)
    def query(self, query, *parameters, **kwparameters):
        """Returns a row list for the given query and parameters."""
        cursor = self._cursor()
        try:
            cursor.execute(query, kwparameters or parameters)
            result = cursor.fetchall()
            return result
        finally:
            cursor.close()
    def get(self, query, *parameters, **kwparameters):
        """Returns the (singular) row returned by the given query.
        """
        cursor = self._cursor()
        try:
            cursor.execute(query, kwparameters or parameters)
            return cursor.fetchone()
        finally:
            cursor.close()
    def execute(self, query, *parameters, **kwparameters):
        """Executes the given query, returning the lastrowid from the query."""
        cursor = self._cursor()
        try:
            cursor.execute(query, kwparameters or parameters)
            return cursor.lastrowid
        except Exception as e:
            if e.args[0] == 1062:
                pass
            else:
                traceback.print_exc()
                raise e
        finally:
            cursor.close()
    insert = execute
    # =============== high level method for table ===================
    def table_has(self, table_name, field, value):
        if isinstance(value, str):
            value = value.encode('utf8')
        sql_str = f'SELECT {field} FROM {table_name} WHERE {field}="{value}"'
        d = self.get(sql_str)
        return d
    def table_insert(self, table_name, item):
        """item is a dict : key is mysql table field"""
        fields = list(item.keys())
        values = list(item.values())
        field_str = ','.join(fields)
        val_str = ','.join(['%s'] * len(item))
        for i in range(len(values)):
            if isinstance(values[i], str):
                values[i] = values[i].encode('utf8')
        sql_str = f'INSERT INTO {table_name} ({field_str}) VALUES({val_str})'
        try:
            last_id = self.execute(sql_str, *values)
            return last_id
        except Exception as e:
            if e.args[0] == 1062:
                # just skip duplicated item
                pass
            else:
                traceback.print_exc()
                print('sql:', sql)
                print('item:')
                for i in range(len(fields)):
                    vs = str(values[i])
                    if len(vs) > 300:
                        print(fields[i], ' : ', len(vs), type(values[i]))
                    else:
                        print(fields[i], ' : ', vs, type(values[i]))
                raise e
    def table_update(self, table_name, updates, field_where, value_where):
        """updates is a dict of {field_update:value_update}"""
        upsets = []
        values = []
        for k, v in updates.items():
            s = '%s=%%s' % k
            upsets.append(s)
            values.append(v)
        upsets = ','.join(upsets)
        sql_str = f'UPDATE {table_name} SET {upsets} WHERE {field_where}="{value_where}"'
        self.execute(sql_str, *values)
if __name__ == '__main__':
    db = Connection(
        'localhost',
        'db_name',
        'user',
        'password'
    )
    # 获取一条记录
    sql = 'select * from test_table where id=%s'
    data = db.get(sql, 2)
    # 获取多天记录
    sql = 'select * from test_table where id>%s'
    data = db.query(sql, 2)
    # 插入一条数据
    sql = 'insert into test_table(title, url) values(%s, %s)'
    last_id = db.execute(sql, 'test', 'http://a.com/')
    # 或者
    last_id = db.insert(sql, 'test', 'http://a.com/')
    # 使用更高级的方法插入一条数据
    item = {
        'title': 'test',
        'url': 'http://a.com/',
    }
    last_id = db.table_insert('test_table', item)

functions.py

# -*- coding: utf-8 -*-
# @Author  : 佛祖保佑, 永无 bug
# @Date    : 
# @File    : functions.py
# @Software: PyCharm
# @description : XXX
import re
import requests
import cchardet
import traceback
import urllib.parse as urlparse
async def fetch(session=None, url=None, headers=None, timeout=9, binary=False):
    _headers = {
        'User-Agent': ('Mozilla/5.0 (compatible; MSIE 9.0; '
                       'Windows NT 6.1; Win64; x64; Trident/5.0)'),
    }
    _headers = headers if headers else _headers
    try:
        async with session.get(url, headers=_headers, timeout=timeout) as response:
            status_code = response.status
            html_bin_or_text = ''
            html_content = await response.read()
            if not binary:
                encoding = cchardet.detect(html_content)['encoding']
                html_bin_or_text = html_content.decode(encoding, errors='ignore')
            request_url = str(response.url)
    except Exception as e:
        msg = 'Failed download: {} | exception: {}, {}'.format(url, str(type(e)), str(e))
        print(msg)
        html_bin_or_text = ''
        status_code = -1
        request_url = url
    return status_code, html_bin_or_text, request_url
def downloader(url=None, timeout=10, headers=None, debug=False, binary=False):
    _headers = {
        'User-Agent': ('Mozilla/5.0 (compatible; MSIE 9.0; '
                       'Windows NT 6.1; Win64; x64; Trident/5.0)'),
    }
    _headers = headers if headers else _headers
    request_url = url
    try:
        r = requests.get(url, headers=_headers, timeout=timeout)
        if binary:
            html_bin_or_text = r.content
        else:
            encoding = cchardet.detect(r.content)['encoding']
            html_bin_or_text = r.content.decode(encoding, errors='ignore')
        status_code = r.status_code
        request_url = r.url
    except BaseException as be:
        if debug:
            traceback.print_exc()
        msg = 'failed download: {}'.format(url)
        print(msg)
        html_bin_or_text = b'' if binary else ''
        status_code = -1
    return status_code, html_bin_or_text, request_url
g_bin_postfix = {
    'exe', 'doc', 'docx', 'xls', 'xlsx', 'ppt', 'pptx', 'pdf',
    'jpg', 'png', 'bmp', 'jpeg', 'gif', 'zip', 'rar', 'tar',
    'bz2', '7z', 'gz', 'flv', 'mp4', 'avi', 'wmv', 'mkv', 'apk'
}
g_news_postfix = ['.html?', '.htm?', '.shtml?', '.shtm?']
def clean_url(url=None):
    # 1. 是否为合法的http url
    if not url.startswith('http'):
        return ''
    # 2. 去掉静态化url后面的参数
    for np in g_news_postfix:
        p = url.find(np)
        if p > -1:
            p = url.find('?')
            url = url[:p]
            return url
    # 3. 不下载二进制类内容的链接
    up = urlparse.urlparse(url)
    path = up.path
    if not path:
        path = '/'
    postfix = path.split('.')[-1].lower()
    if postfix in g_bin_postfix:
        return ''
    # 4. 去掉标识流量来源的参数
    # badquery = ['spm', 'utm_source', 'utm_source', 'utm_medium', 'utm_campaign']
    good_queries = []
    for query in up.query.split('&'):
        qv = query.split('=')
        if qv[0].startswith('spm') or qv[0].startswith('utm_'):
            continue
        if len(qv) == 1:
            continue
        good_queries.append(query)
    query = '&'.join(good_queries)
    url = urlparse.urlunparse((
        up.scheme,
        up.netloc,
        path,
        up.params,
        query,
        ''  # crawler do not care fragment
    ))
    return url
g_pattern_tag_a = re.compile(r'<a[^>]*?href=[\'"]?([^> \'"]+)[^>]*?>(.*?)</a>', re.I | re.S | re.M)
def extract_links_re(url=None, html=None):
    """use re module to extract links from html"""
    news_links = set()
    tag_a_list = g_pattern_tag_a.findall(html)
    for tag_a in tag_a_list:
        link = tag_a[0].strip()
        if not link:
            continue
        link = urlparse.urljoin(url, link)
        link = clean_url(link)
        if not link:
            continue
        news_links.add(link)
    return news_links
def init_file_logger(f_name=None):
    # config logging
    import logging
    from logging.handlers import TimedRotatingFileHandler
    ch = TimedRotatingFileHandler(f_name, when="midnight")
    ch.setLevel(logging.INFO)
    # create formatter
    fmt = '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    formatter = logging.Formatter(fmt)
    # add formatter to ch
    ch.setFormatter(formatter)
    logger = logging.getLogger(f_name)
    # add ch to logger
    logger.addHandler(ch)
    return logger
if __name__ == '__main__':
    temp_url = 'http://news.baidu.com/'
    t_status_code, t_html, t_url = downloader(url=temp_url)
    print(f'[{t_status_code}, {t_url}]:{len(t_html)}')

config.py

db_host = 'localhost'
db_db = 'crawler'
db_user = 'your-user'
db_password = 'your-password'

新闻爬虫 ( 同步 )

news_sync.py

# -*- coding: utf-8 -*-
# @Author  : 佛祖保佑, 永无 bug
# @Date    : 
# @File    : news_sync.py
# @Software: PyCharm
# @description : XXX
import urllib.parse as urlparse
import lzma
import farmhash
import traceback
from ezpymysql import Connection
from url_pool import UrlPool
import functions as fn
import config
class NewsCrawlerSync:
    def __init__(self, name):
        self.db = Connection(
            config.db_host,
            config.db_db,
            config.db_user,
            config.db_password
        )
        self.logger = fn.init_file_logger(name + '.log')
        self.url_pool = UrlPool(name)
        self.hub_hosts = None
        self.load_hubs()
    def load_hubs(self,):
        sql = 'select url from crawler_hub'
        data = self.db.query(sql)
        self.hub_hosts = set()
        hubs = []
        for d in data:
            host = urlparse.urlparse(d['url']).netloc
            self.hub_hosts.add(host)
            hubs.append(d['url'])
        self.url_pool.set_hubs(hubs, 300)
    def save_to_db(self, url, html):
        url_hash = farmhash.hash64(url)
        sql = f'select url from crawler_html where urlhash={url_hash}'
        d = self.db.get(sql, url_hash)
        if d:
            if d['url'] != url:
                msg = 'farm_hash collision: %s <=> %s' % (url, d['url'])
                self.logger.error(msg)
            return True
        if isinstance(html, str):
            html = html.encode('utf8')
        html_lzma = lzma.compress(html)
        sql = 'insert into crawler_html(urlhash, url, html_lzma) values(%s, %s, %s)'
        good = False
        try:
            self.db.execute(sql, url_hash, url, html_lzma)
            good = True
        except Exception as e:
            if e.args[0] == 1062:
                # Duplicate entry
                good = True
                pass
            else:
                traceback.print_exc()
                raise e
        return good
    def filter_good(self, urls):
        good_links = []
        for url in urls:
            host = urlparse.urlparse(url).netloc
            if host in self.hub_hosts:
                good_links.append(url)
        return good_links
    def process(self, url, is_hub):
        status, html, redirected_url = fn.downloader(url)
        self.url_pool.set_status(url, status)
        if redirected_url != url:
            self.url_pool.set_status(redirected_url, status)
        # 提取hub网页中的链接, 新闻网页中也有“相关新闻”的链接，按需提取
        if status != 200:
            return
        if is_hub:
            new_links = fn.extract_links_re(redirected_url, html)
            good_links = self.filter_good(new_links)
            print(f"{len(good_links)} / {len(new_links)}, good_links/new_links")
            self.url_pool.add_many(good_links)
        else:
            self.save_to_db(redirected_url, html)
    def run(self,):
        while 1:
            urls = self.url_pool.pop(5)
            for url, is_hub in urls.items():
                self.process(url, is_hub)
if __name__ == '__main__':
    crawler = NewsCrawlerSync('sync_spider')
    crawler.run()

新闻爬虫 ( 异步 )

news_async.py

# -*- coding: utf-8 -*-
# @Author  : 佛祖保佑, 永无 bug
# @Date    : 
# @File    : news_async.py
# @Software: PyCharm
# @description : XXX
import traceback
import time
import asyncio
import aiohttp
import urllib.parse as urlparse
import farmhash
import lzma
# import uvloop
# asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
import sanicdb
from url_pool import UrlPool
import functions as fn
import config
class NewsCrawlerAsync:
    def __init__(self, name):
        self.hub_hosts = set()
        self._workers = 0
        self._workers_max = 30
        self.logger = fn.init_file_logger(name + '.log')
        self.url_pool = UrlPool(name)
        self.loop = asyncio.get_event_loop()
        self.session = aiohttp.ClientSession(loop=self.loop)
        self.db = sanicdb.SanicDB(
            config.db_host,
            config.db_db,
            config.db_user,
            config.db_password,
            loop=self.loop
        )
    async def load_hubs(self, ):
        sql = 'select url from crawler_hub'
        data = await self.db.query(sql)
        hubs = []
        for d in data:
            host = urlparse.urlparse(d['url']).netloc
            self.hub_hosts.add(host)
            hubs.append(d['url'])
        self.url_pool.set_hubs(hubs, 300)
    async def save_to_db(self, url, html):
        url_hash = farmhash.hash64(url)
        sql = 'select url from crawler_html where urlhash=%s'
        d = await self.db.get(sql, url_hash)
        if d:
            if d['url'] != url:
                msg = 'farmhash collision: %s <=> %s' % (url, d['url'])
                self.logger.error(msg)
            return True
        if isinstance(html, str):
            html = html.encode('utf8')
        html_lzma = lzma.compress(html)
        sql = 'insert into crawler_html(urlhash, url, html_lzma) values(%s, %s, %s)'
        good = False
        try:
            await self.db.execute(sql, url_hash, url, html_lzma)
            good = True
        except Exception as e:
            if e.args[0] == 1062:
                # Duplicate entry
                good = True
                pass
            else:
                traceback.print_exc()
                raise e
        return good
    def filter_good(self, urls):
        good_links = []
        for url in urls:
            host = urlparse.urlparse(url).netloc
            if host in self.hub_hosts:
                good_links.append(url)
        return good_links
    async def process(self, url, is_hub):
        status, html, redirected_url = await fn.fetch(self.session, url)
        self.url_pool.set_status(url, status)
        if redirected_url != url:
            self.url_pool.set_status(redirected_url, status)
        # 提取hub网页中的链接, 新闻网页中也有“相关新闻”的链接，按需提取
        if status != 200:
            self._workers -= 1
            return
        if is_hub:
            new_links = fn.extract_links_re(redirected_url, html)
            good_links = self.filter_good(new_links)
            print(f"{len(good_links)} / {len(new_links)}, good_links / new_links")
            self.url_pool.add_many(good_links)
        else:
            await self.save_to_db(redirected_url, html)
        self._workers -= 1
    async def loop_crawl(self):
        await self.load_hubs()
        last_rating_time = time.time()
        counter = 0
        while 1:
            to_pop = self._workers_max - self._workers
            tasks = self.url_pool.pop(to_pop)
            if not tasks:
                print('no url to crawl, sleep')
                await asyncio.sleep(3)
                continue
            for url, is_hub in tasks.items():
                self._workers += 1
                counter += 1
                print('crawl:', url)
                asyncio.ensure_future(self.process(url, is_hub))
            gap = time.time() - last_rating_time
            if gap > 5:
                rate = counter / gap
                print(f'\tloop_crawl() rate:{round(rate, 2)}, counter: {counter}, workers: {self._workers}')
                last_rating_time = time.time()
                counter = 0
            if self._workers > self._workers_max:
                print('====== got workers_max, sleep 3 sec to next worker =====')
                await asyncio.sleep(3)
    def run(self):
        try:
            self.loop.run_until_complete(self.loop_crawl())
        except KeyboardInterrupt:
            print('stopped by yourself!')
            del self.url_pool
            pass
if __name__ == '__main__':
    nc = NewsCrawlerAsync('async_spider')
    nc.run()

分布式爬虫 （ CS 模型 ）

server.py

# -*- coding: utf-8 -*-
# @Author  : 佛祖保佑, 永无 bug
# @Date    : 
# @File    : server.py
# @Software: PyCharm
# @description : XXX
from sanic import Sanic
from sanic import response
from my_url_pool import UrlPool
url_pool = UrlPool(__file__)
# 初始化 url_pool，根据你的需要进行修改
hub_urls = []
url_pool.set_hubs(hub_urls, 300)
url_pool.add('https://news.sina.com.cn/')
# init
main_app = Sanic(__name__)
@main_app.listener('after_server_stop')
async def cache_url_pool(app=None, loop=None):
    global url_pool
    print('caching url_pool after_server_stop')
    del url_pool
    print('bye!')
@main_app.route('/task')
async def task_get(request=None):
    count = request.args.get('count', 10)
    try:
        count = int(count)
    except BaseException as be:
        count = 10
    urls = url_pool.pop(count)
    return response.json(urls)
@main_app.route('/task', methods=['POST', ])
async def task_post(request=None):
    result = request.json
    url_pool.set_status(result['url'], result['status'])
    if result['url_real'] != result['url']:
        url_pool.set_status(result['url_real'], result['status'])
    if result['new_urls']:
        print('receive URLs:', len(result['new_urls']))
        for url in result['new_urls']:
            url_pool.add(url)
    return response.text('ok')
if __name__ == '__main__':
    main_app.run(host='0.0.0.0', port=8080, debug=False, access_log=False, workers=1)
    pass

client.py

# -*- coding: utf-8 -*-
# @Author  : 佛祖保佑, 永无 bug
# @Date    : 
# @File    : client.py
# @Software: PyCharm
# @description : XXX
import re
import cchardet
import traceback
import time
import json
import asyncio
import urllib.parse as urlparse
import aiohttp
# import uvloop
# asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
p_tag_a = re.compile(r'<a[^>]*?href=[\'"]?([^> \'"]+)[^>]*?>(.*?)</a>', re.I | re.S | re.M)
def extract_links_re(url, html):
    new_links = set()
    aa = p_tag_a.findall(html)
    for a in aa:
        link = a[0].strip()
        if not link:
            continue
        link = urlparse.urljoin(url, link)
        if not link.startswith('http'):
            continue
        new_links.add(link)
    return new_links
class CrawlerClient:
    def __init__(self, ):
        self._workers = 0
        self.workers_max = 20
        self.server_host = 'localhost'
        self.server_port = 8080
        self.headers = {'User-Agent': ('Mozilla/5.0 (compatible; MSIE 9.0; '
                                       'Windows NT 6.1; Win64; x64; Trident/5.0)')}
        self.loop = asyncio.get_event_loop()
        self.queue = asyncio.Queue(loop=self.loop)
        self.session = aiohttp.ClientSession(loop=self.loop)
    async def download(self, url, timeout=25):
        status_code = 900
        html = ''
        url_now = url
        try:
            async with self.session.get(url_now, headers=self.headers, timeout=timeout) as response:
                status_code = response.status
                html = await response.read()
                encoding = cchardet.detect(html)['encoding']
                html = html.decode(encoding, errors='ignore')
                url_now = str(response.url)
        except BaseException as be:
            # traceback.print_exc()
            print('=== exception: ', be, type(be), str(be))
            msg = 'Failed download: {} | exception: {}, {}'.format(url, str(type(be)), str(be))
            print(msg)
        return status_code, html, url_now
    async def get_urls(self, ):
        count = self.workers_max - self.queue.qsize()
        if count <= 0:
            print('no need to get urls this time')
            return None
        url = f'http://{self.server_host}:{self.server_port}/task?count={count}'
        try:
            async with self.session.get(url, timeout=3) as response:
                if response.status not in [200, 201]:
                    return
                jsn = await response.text()
                urls = json.loads(jsn)
                msg = f'get_urls()  to get [{count}] but got[{len(urls)}], @{time.strftime("%Y-%m-%d %H:%M:%S")}'
                print(msg)
                for kv in urls.items():
                    await self.queue.put(kv)
                print('queue size:', self.queue.qsize(), ', _workers:', self._workers)
        except BaseException as be:
            traceback.print_exc()
            return
    async def send_result(self, result):
        url = f'http://{self.server_host}:{self.server_port}/task'
        try:
            async with self.session.post(url, json=result, timeout=3) as response:
                return response.status
        except BaseException as be:
            traceback.print_exc()
            pass
    @staticmethod
    def save_html(url, html):
        print('saved:', url, len(html))
    @staticmethod
    def filter_good(urls):
        """根据抓取目的过滤提取的URLs，只要你想要的"""
        good = []
        for url in urls:
            if url.startswith('http'):
                good.append(url)
        return good
    async def process(self, url, is_hub):
        status, html, url_now = await self.download(url)
        self._workers -= 1
        print('downloaded:', url, ', html:', len(html))
        if html:
            new_urls = extract_links_re(url, html)
            new_urls = self.filter_good(new_urls)
            self.save_html(url, html)
        else:
            new_urls = []
        result = {
            'url': url,
            'url_real': url_now,
            'status': status,
            'new_urls': new_urls,
        }
        await self.send_result(result)
    async def loop_get_urls(self, ):
        print('loop_get_urls() start')
        while 1:
            await self.get_urls()
            await asyncio.sleep(1)
    async def loop_crawl(self, ):
        print('loop_crawl() start')
        asyncio.ensure_future(self.loop_get_urls())
        counter = 0
        while 1:
            item = await self.queue.get()
            url, url_level = item
            self._workers += 1
            counter += 1
            asyncio.ensure_future(self.process(url, url_level))
            if self._workers > self.workers_max:
                print('====== got workers_max, sleep 3 sec to next worker =====')
                await asyncio.sleep(3)
    def start(self):
        try:
            self.loop.run_until_complete(self.loop_crawl())
        except KeyboardInterrupt:
            print('stopped by yourself!')
            pass
def run():
    ant = CrawlerClient()
    ant.start()
if __name__ == '__main__':
    run()