Как расширить класс Thread в Python - подробно на примерах

В этом уроке вы узнаете, как расширить класс Thread в Python для запуска кода в новом потоке.

Введение в класс потоков Python

Когда программа Python запускается, у нее есть поток, называемый основным потоком. Иногда вам нужно перенести задачи, связанные с вводом-выводом, в новый поток, чтобы выполнять их одновременно. Для этого вы используете встроенный модуль threading.

Один из способов выполнить код в новом потоке — расширить класс Thread модуля threading:

Сначала определите подкласс класса threading.Thread.
Во-вторых, переопределите метод __init__(self, [,args]) внутри метода __init__() подкласса, чтобы добавить собственные аргументы.
В-третьих, переопределите метод run(self, [,args]) внутри подкласса, чтобы настроить поведение нового класса потока при создании нового потока.

Пример расширения класса Thread

Давайте рассмотрим пример расширения класса Thread. Мы разработаем класс, который выполняет HTTP-запрос к URL-адресу и отображает код ответа:

class HttpRequestThread(Thread):
    def __init__(self, url: str) -> None:
        super().__init__()
        self.url = url

    def run(self) -> None:
        print(f'Checking {self.url} ...')
        try:
            response = urllib.request.urlopen(self.url)
            print(response.code)
        except urllib.error.HTTPError as e:
            print(e.code)
        except urllib.error.URLError as e:
            print(e.reason)

Как это работает.

Сначала определите HttpRequestThread, который расширяет класс Thread из модуля threading:

class HttpRequestThread(Thread):

Во-вторых, определите метод __init__(), который принимает URL-адрес. Внутри метода __init__() вызывается метод __init__() суперкласса.

def __init__(self, url: str) -> None:
    super().__init__()
    self.url = url

В-третьих, переопределите метод run, который использует urllib для получения кода состояния HTTP указанного URL-адреса и отображения его на консоли:

def run(self) -> None:
    print(f'Checking {self.url} ...')
    try:
        response = urllib.request.urlopen(self.url)
        print(response.code)
    except urllib.error.HTTPError as e:
        print(e.code)
    except urllib.error.URLError as e:
        print(e.reason)

Чтобы использовать класс HttpRequestThread, вы создаете экземпляры класса HttpRequestThread и вызываете метод start(). Кроме того, вы можете вызвать метод join(), чтобы дождаться завершения всех потоков.

Ниже определяется функция main(), использующая класс HttpRequestThread:

def main() -> None:
    urls = [
        'https://httpstat.us/200',
        'https://httpstat.us/400'
    ]

    threads = [HttpRequestThread(url) for url in urls]

    [t.start() for t in threads]

    [t.join() for t in threads]

Как это работает:

Сначала определим список URL-адресов, которые мы хотим проверить:

urls = [
    'https://httpstat.us/200',
    'https://httpstat.us/400'
]

Затем создадим экземпляры HttpRequestThread на основе списка URL-адресов, используя понимание списка. Генератор списка возвращает список экземпляров класса HttpRequestThread:

threads = [HttpRequestThread(url) for url in urls]

В-третьих, вызовем метод start() каждого потока в списке потоков:

[t.start() for t in threads]

Наконец, вызовем соединение каждого экземпляра потока, чтобы дождаться завершения всех потоков:

[t.join() for t in threads]

Положим все это вместе:

from threading import Thread
import urllib.request


class HttpRequestThread(Thread):
    def __init__(self, url: str) -> None:
        super().__init__()
        self.url = url

    def run(self) -> None:
        print(f'Checking {self.url} ...')
        try:
            response = urllib.request.urlopen(self.url)
            print(response.code)
        except urllib.error.HTTPError as e:
            print(e.code)
        except urllib.error.URLError as e:
            print(e.reason)


def main() -> None:
    urls = [
        'https://httpstat.us/200',
        'https://httpstat.us/400'
    ]

    threads = [HttpRequestThread(url) for url in urls]

    [t.start() for t in threads]

    [t.join() for t in threads]


if __name__ == '__main__':
    main()

Выход:

Checking https://httpstat.us/200 ...
Checking https://httpstat.us/400 ...
200
400