Crawler Notes: Proxy Pool and Thread Pool

Preface

Crawlers and anti-crawlers are a spear and a shield. A very common anti-crawler method is banning IPs. If an IP makes frequent requests in a short time, it can be rate-limited or added to a blacklist. I’ve also covered this in my previous backend development related blogs.

However, today we’re talking about crawlers, so the countermeasure is to use a proxy pool—just use a different IP for each request, plus UA simulation, which is completely normal user behavior and can avoid rate-limiting and blacklist anti-crawling.

Then crawlers are an IO-intensive type of program. If executed entirely in a single thread it will be very slow, so you can use multithreading to improve data collection efficiency. But managing multithreading yourself is too much trouble, so I chose a thread pool~

Proxy Pool

A complete proxy pool should be able to implement the following features

• Batch collect proxies (or import proxies we purchased via an API; but for occasional use, free ones are fine)
• Automatically verify proxy validity after collection
• Store valid proxies
• Provide an API to get a random proxy
• Provide APIs to manage (delete, add) proxies

Reinventing the wheel is too much trouble. Isn’t the whole point of using Python “life is short, I use Python”? And the community hasn’t let us down—there are many good open-source Python proxy pool projects. Here I picked a project with 14k+ Stars on GitHub to use, named ProxyPool.

After trying it out, it’s pretty good!

Of course there are many other thread pool projects as well; I haven’t tested them. Interested readers can check the first link in the references.

Deployment and Running

Project address: https://github.com/jhao104/proxy_pool

The official docs provide two deployment methods, including downloading the code to run it and Docker. Since Docker is available, of course I’ll pick the most convenient option—Docker!

But the official Docker command isn’t convenient enough, because this proxy pool also depends on a Redis service. Here I wrote a docker-compose config to use:

version: "3"
services:
  redis:
    image: redis
    expose:
      - 6379

  web:
    restart: always
    image: jhao104/proxy_pool
    environment:
      - DB_CONN=redis://redis:6379/0
    ports:
      - "5010:5010"
    depends_on:
      - redis

Save it in some folder, then run the command to start the Docker containers

docker-compose up

Here I configured the port as 5010, same as the official site. Anyone who needs to can modify it themselves~

After the project is up and running, visit http://127.0.0.1:5010 in the browser to get all the APIs. Each API is easy to understand from its name.

{
  "url": [
    {
      "desc": "get a proxy",
      "params": "type: ''https'|''",
      "url": "/get"
    },
    {
      "desc": "get and delete a proxy",
      "params": "",
      "url": "/pop"
    },
    {
      "desc": "delete an unable proxy",
      "params": "proxy: 'e.g. 127.0.0.1:8080'",
      "url": "/delete"
    },
    {
      "desc": "get all proxy from proxy pool",
      "params": "type: ''https'|''",
      "url": "/all"
    },
    {
      "desc": "return proxy count",
      "params": "",
      "url": "/count"
    }
  ]
}

Using It in Code

Since this proxy pool provides HTTP APIs, in theory it can be used with any language

Here I’ll write it in Python

Getting a Random Proxy

Here I wrote two methods that wrap the operations of getting a random proxy and deleting a proxy

import requests

PROXY_POOL_URL = 'http://127.0.0.1:5010'

def get_proxy():
    proxy = requests.get(f"{PROXY_POOL_URL}/get/").json().get("proxy")
    return {'http': proxy, 'https': proxy}

def delete_proxy(proxy):
    requests.get(f"{PROXY_POOL_URL}/delete/?proxy={proxy}")

Getting a Random Header

Use the fake_useragent library to generate a random UserAgent, simulating requests from different user browsers

from fake_useragent import UserAgent

def get_header():
    return {
        "Accept": "application/json, text/plain, */*",
        "Accept-Encoding": "gzip, deflate",
        "Accept-Language": "zh-CN,zh;q=0.9,en-US;q=0.8,en;q=0.7,th;q=0.6",
        "Cache-Control": "no-cache",
        "Connection": "keep-alive",
        "Pragma": "no-cache",
        "User-Agent": ua.random
    }

Wrapping Network Requests

Because we didn’t buy paid proxies, we’re using free proxies automatically collected by the proxy pool. As everyone knows, the quality of free proxies isn’t guaranteed, so I wrote a retry feature. After failures exceed the max retry count, delete this proxy and switch to another one to try again~

The max retry count can be configured via the MAX_RETRY_COUNT variable

MAX_RETRY_COUNT = 5

def request_get(url) -> Tuple[Response, str]:
    retry_count = 1
    proxy = get_proxy()
    while retry_count <= MAX_RETRY_COUNT:
        logger.debug(f'第{retry_count}次请求 - 网址 {url} - 代理 {proxy.get("http")}')
        try:
            resp = requests.get(url, proxies=proxy, headers=get_header(), timeout=15)
            return resp, proxy.get('http')
        except Exception:
            logger.error(f'请求失败 - 网址 {url}')
            retry_count += 1
    # 删除代理池中代理
    logger.warning(f'全部{MAX_RETRY_COUNT}次请求都失败 - 删除代理 {proxy.get("http")}')
    delete_proxy(proxy.get('http'))
    return request_get(url)

This function returns a tuple of type (Response, str). Considering that the data format obtained from different requests may vary, I didn’t use resp.json() or resp.text. You can process the data yourself after calling this function.

It also returns a proxy server address of type str, in ip:port form.

The way to call it is of this form: resp, proxy = request_get(url)

Because this request_get function I wrapped is only the most basic data fetching, but the data you get isn’t necessarily correct and usable. For example, if rate limiting or a blacklist is triggered, the data you get will be empty. In that case, after calling this function to get the data you can add a check: if this proxy IP has already been banned, you can call the delete_proxy method to delete that proxy.

Thread Pool

Crawlers are an IO-intensive type of program. If executed entirely in a single thread it will be very slow, so you can use multithreading to improve data collection efficiency. But managing multithreading yourself is too much trouble, so I chose a thread pool~

A thread pool is a group of pre-instantiated idle threads, ready to accept work. Creating a new thread object for each task to be executed asynchronously is expensive. With a thread pool, you can add tasks to a task queue, and the thread pool assigns an available thread to the task. Thread pools help avoid creating or destroying unnecessary threads.

Previously I used the threadpool pip package to implement a thread pool. It felt pretty good, but when used for crawling there was a chance of unexplained freezes. I don’t know where the problem was. Later I saw online materials saying this threadpool is more suitable for CPU-intensive operations…

PS: I looked at threadpool’s source implementation. Impressive—421 lines of code and it implements thread pool functionality~

And it’s based on the threading module. Nice.

This time I switched to the thread pool implementation that comes with Python’s standard library. In fact, there are two kinds of “pools” in Python

• multiprocessing.Pool
• multiprocessing.pool.Threadpool

The similarities and differences between these two:

multiprocessing.pool.ThreadPool behaves the same way as multiprocessing.Pool. The difference is that multiprocessing.pool.Threadpool uses threads to run the worker logic, while multiprocessing.Pool uses worker processes.

But I won’t use either of these for now, because there’s a better choice.

Starting from Python3.2, the standard library provides us with the concurrent.futures module, which provides two classes, ThreadPoolExecutor and ProcessPoolExecutor, implementing a further abstraction over threading and multiprocessing (here we mainly focus on the thread pool). Not only can it help us automatically schedule threads, it can also:

1. The main thread can get the status of a certain thread (or task), as well as the return value.
2. When a thread completes, the main thread can know immediately.
3. Make the coding interfaces for multithreading and multiprocessing consistent.

So let’s look at the code

Code

Simple usage

def crawl_data(page):
    ...

from concurrent.futures import ThreadPoolExecutor, wait, ALL_COMPLETED
pool = ThreadPoolExecutor(8)
logger.info('线程池启动')
tasks = [pool.submit(crawl_data, page) for page in range(1, 100)]
wait(tasks, return_when=ALL_COMPLETED)
logger.info('线程池结束')

Analysis of the code above:

• The crawl_data function is the crawler function; the specific code is omitted
• ThreadPoolExecutor(8) means creating a thread pool with 8 threads in parallel
• Then using a list comprehension, the pool.submit method is used to add tasks to the thread pool
• The wait function is used to wait for the thread pool to finish executing.

Besides the pool.submit method, it also supports the map method to add tasks in batch

Usage is as follows:

pool = ThreadPoolExecutor(8)
pool.map(crawl_data, range(1,100))

The second argument of the map method is the list of arguments to pass to the task, so however many arguments are in the list, that many tasks are created~

After testing it’s very stable, hahaha. Standard library stuff is still the most reliable~

References

• https://suyin-blog.club/2021/2G4HXBY/#proxy-pool-%E6%8E%A8%E8%8D%90
• The past and present of python threadpool: https://zhangchenchen.github.io/2017/05/18/python-thread-pool/
• https://www.delftstack.com/zh/howto/python/python-threadpool-differences/
• [python] ThreadPoolExecutor thread pool: https://www.jianshu.com/p/b9b3d66aa0be