что такое worker python
Многопоточный Python на примерах: как правильно хранить настройки приложения
Если опустить первое и самое главное предубеждение относительно питонячьей многопоточности у большинства программистов — что её не существует из-за GIL, — то остается другое, и, наверное, вполне достоверное: многопоточность — это сложно, и нам этого, пожалуйста, не надо. И знаете что? Так оно и есть. Многопоточность — это сложно, особенно когда выбираешься за пределы стандартных руководств и попадаешь со своей многопоточной поделкой в реальный мир. И, возможно, вам не нужно. Ни здесь, ни далее я не буду обсуждать целесообразность написания многопоточного кода на Python и сразу перейду к тому, как это делать.
Если эта статья всем понравится, за ней может последовать серия на ту же тему: как писать многопоточный код с примерами на Python. Конкретно в этой статье примеры будут взяты из фреймворка Polog, который я пишу в свободное время. Он предназначен для логирования, но я постарался вынести все базовые принципы и паттерны в эту статью в специфико-агностичной форме. А там, где без специфики фреймворка не обойтись, буду указывать, в чём именно она состоит. Хочется верить, что это даже поспособствует усвоению материала за счёт каких-то ассоциативных связей, что ли.
Дисклеймер
Несмотря на то, что я привожу собственный код в качестве примера, я тем самым не утверждаю, что он написан лучшим из возможным способов. Более того, я даже не уверен, что вы не сможете найти в нём каких-либо (возможно, фатальных) недостатков. Если вдруг вы в состоянии это сделать, было бы круто, не сдерживайте себя.
Кроме того, здесь не будет ликбеза про многопоточность. Предполагается, что вы уже прочитали что-то вроде документации к модулю threading из стандартной библиотеки, и в целом хотя бы немного знакомы с темой. Вам не нужно объяснять, чем потоки отличаются от вашей бабушки, кто такие эти наши локи (и каковы их семейные взаимоотношения с дедом), что такое состояние гонки и всякое подобное. Про это всё вы уже сами прочитали и теперь вам интересно не столько то, что такое многопоточность, сколько то, как её применение может выглядеть в конкретном проекте.
Чего мы хотим?
Существует самый простой и очевидный способ хранить настройки приложения в качестве обычных переменных внутри модуля или класса. Его используют в том числе крупные веб-фреймворки, и я поначалу делал так же. Так почему же по итогу пришлось всё переделывать? Дело в том, что к хранению настроек внутри фреймворка возникли довольно специфические требования:
Прежде всего, необходима гибкая валидация значений. Причём, поскольку это библиотека, пользователь должен получать информативное исключение с описанием того, что именно он сделал не так, когда попытался применить заведомо невозможную настройку (например, установить какой-нибудь интервал времени в виде отрицательной величины).
Необходимо валидировать не просто значения сами по себе, но и их комбинации. К примеру, в Polog поддерживается два типа движков: синхронный (без использования воркеров в отдельных потоках) и асинхронный. Синхронный подгружается, когда настройка pool_size установлена в значение 0, асинхронный — при любом значении больше нуля. Асинхронный движок работает по схеме producer–consumer с передачей данных через очередь. Так вот, для этой самой очереди можно установить лимит — целое число больше нуля, а можно не устанавливать, оставив значение 0 по умолчанию. «Невозможной» комбинацией является pool_size равный нолю, и при этом max_queue_size (лимит размера очереди) больше нуля, поскольку никакой очереди в синхронном движке просто нет. Хотя каждое отдельное из значений — нулевой размер пула и ненулевой размер очереди — само по себе валидно, при попытке установить такую комбинацию мы тоже должны кинуть исключение с описанием конфликтующих полей.
Нужна возможность изменять настройки в любой момент. Исходно всё работало так: перед записью самого первого лога вы указывали нужные значения для всех пунктов настроек, и потом, когда первый лог создавался, инициализировались все внутренние объекты фреймворка согласно настройкам на этот момент времени. Дальше уже ничего поменять было нельзя. Так быть не должно, в идеале любой пункт настроек должно быть возможно изменить в любой момент времени, не потеряв при этом ни одной записи. Дальше станет ясно, почему этого добиться не так уж просто.
Каждый пункт настроек должен иметь значение по умолчанию.
Всё должно быть защищено от состояния гонки.
Хранилище должно иметь интерфейс словаря. Это просто и удобно, а ещё так компоненты фреймворка проще тестировать, подменяя хранилище настроек словарем.
Базовая концепция
У нас есть класс, оборачивающий доступ к словарю, объявленному как один из его атрибутов. Примерно вот такой:
Ключи в этом словаре — имена пунктов настроек, а значения — объекты, в которых хранятся значения. Именно в этих объектах класса SettingPoint и происходит почти вся описанная выше магия: валидация значений с киданием красивых исключений, блокировки и всё прочее. Далее мы сосредоточимся на устройстве и механизмах внутри этого класса.
Валидируем одиночные значения
Рассмотрение класса SettingPoint мы начнём с самого простого — как валидируются значения. Напомню, что создание экземпляра выглядело примерно так:
На этом этапе нас интересует словарь, который мы передаем для инициализации экземпляра как аргумент proves :
Ищем конфликты
Немного усложняем задачу: теперь нам нужно проверить, что два разных значения настроек не образуют невозможную комбинацию, хотя каждое из них в отдельности валидно. Для этого при инициализации объекта SettingPoint мы передаём словарь с описанием конфликтов, примерно такой:
Ключи в этом словаре — это названия полей, конфликты с которыми мы проверяем, а значения — функции. При каждой попытке сохранить новое значение этого пункта настроек мы передаём в каждую из функций три параметра: новое значение; значение, которое было до этого; и текущее значение того поля, конфликты с которым мы ищем. Для этого в объекте поля запускается простенький метод:
Что ж, мы провели два типа проверок и теперь должны быть уверены, что новое значение настройки нам подходит. Теперь наша задача — применить полученные данные.
Меняем настройки в рантайме
Пунктов настроек больше одного, но я опишу конвейер изменения одного из них, с которым пришлось повозиться больше всех. Чтобы объяснить, почему так сложно его изменить, придётся немного погрузить в устройство фреймворка до и после. Будет немного специфики, без неё никуда.
Исходно ядро Polog было исключительно многопоточным, на базе самописного thread pool. Тредпул работал по паттерну издатель-подписчик, то есть состоял из тр`х логических частей:
Объект, который передаёт логи в очередь.
Набор потоков-воркеров. В каждом из них запущен бесконечный цикл, в котором они ожидают появления новых записей в очереди. Как только первый из них забирает лог, он сразу его как-то обрабатывает, например, записывает в файл.
Воркеры создавались «лениво», в момент записи первого лога. Это позволяло применить любые настройки на этапе инициализации программы. Проблема только в том, что тредпул получался «приколоченным гвоздями». После записи первого лога уже ни добавить воркеров, ни убрать лишних. Кроме того, не была предусмотрена возможность использования беспоточного движка: класс движка тоже, получается, приколочен. Валидным размером тредпула в настройках было только число не меньше единицы.
Существует две базовые стратегии изменения тредпула «на лету»:
Изменение текущего тредпула с помощью добавления новых или удаления из него старых воркеров. Это вполне легко реализовать, нужно лишь получить новое число потоков; сравнить его с текущим; определить, добавляем или убираем; и применить изменение.
Обернуть компонент с тредпулом в объект уровнем выше, который при каждом изменении настроек будет пересоздавать пул с нуля.
Казалось бы, первый вариант экономичнее. Надо нам добавить 101-й поток, мы не занимаемся бессмысленным уничтожением и созданием заново предыдущей сотни. Мы просто создаём ещё один и добавляем в пул. Однако я выбрал второй вариант, поскольку его API универсальнее и позволяет легко изменять любые характеристики движка, а не только количество воркеров. Любое изменение — это уничтожение предыдущего компонента и создание нового в соответствии с новыми настройками. У такого решения, кстати, был важный и полезный побочный эффект: стало возможным существование более одного типа движков. Старый уничтожается, а новый может быть каким угодно, он не связан со старым какими-либо общими механизмами.
«Настоящих» движков (не обёрток) в настоящее время доступно два типа: синхронный и асинхронный, как я уже писал выше. Их порождает вот такая простенькая функция:
Если размер запрошенного пула потоков (настройка pool_size ) больше нуля, то создаётся асинхронный движок на базе тредпула. А если равен нулю, то создаётся более примитивный синхронный движок, у которого нет тредпула, а есть просто метод, куда можно передать лог и он будет обработан.
Для понимания, как выглядит минимальный возможный движок, синхронная версия выглядит так:
Это всё. Метод самоуничтожения спрятан в классе, от которого наследуется движок, и не делает ничего, поскольку тут нет никаких ресурсов, которые надо освобождать.
Асинхронный немного сложнее, ведь он должен уметь завершаться, не потеряв ни одного лога в очереди. Давайте на него тоже глянем, прежде чем идти дальше:
Как видите, вся работа по корректному завершению происходит внутри пула, так что провалимся в него:
Итак, завершение работы пула происходит в три этапа:
Дожидаемся, пока очередь с логами опустеет. При этом важно, чтобы с другой стороны её никто не пополнял, но об этом будет дальше. Ожидание опустения очереди — это обычный цикл, в котором с определённой периодичностью запрашивается размер очереди, и который завершается, когда очередь пуста.
Сообщаем каждому воркеру, что пора заканчивать работу. Дело в том, что у него на руках в этот момент ещё может быть последний взятый из очереди лог, поэтому мы не можем прервать его насильно и вынуждены вежливо попросить. Воркер, в свою очередь (на это мы посмотрим дальше), получив такое сообщение, должен закончить с последним взятым из очереди логом, после чего добровольно суициднуться прервать собственный цикл.
Джойним каждый поток с воркером.
Внутри воркера обработка сообщения об остановке выглядит как установка в значение True флага self.stopped :
Этот пример кода, возможно, придётся пораскуривать некоторое время, мне и самому его не так просто в голове уместить. Прежде всего требует пояснений, почему здесь два бесконечных цикла, а не один, как я обещал ранее. Дело в том, что если воркер ожидает из очереди данных без таймаута, то возможна ситуация, когда мы дождались бы опустения очереди, проставили флаг завершения и дальше бесконечно ждали бы его. Поэтому воркер ожидает данные из очереди не бесконечно, а периодически «просыпается», чтобы проверить, не попросили ли его прерваться. Не попросили — прекрасно, возвращаемся в очередь.
Выше мы выяснили, как происходит рождение и остановка движков, однако здесь мне придётся вернуться немного назад, чтобы рассказать ещё о паре важных моментов.
Во-вторых, при подмене движка во время работы программы нужно учитывать, что всё это время его обёртку могут вызывать логгеры из разных потоков, поэтому обёртку движка мы также временно блокируем. Код ниже демонстрирует, как происходит блокировка и последующая разблокировка движка при перезагрузке:
Таким образом, любой поток, который попытается записать лог в тот момент, пока движок перезагружается, будет временно заблокирован, и продолжит свое выполнение, когда всё уже завершится, ничего даже не заметив. Лог будет записан как положено и не потеряется.
Если вы добрались до этого места и всё ещё ничего не поняли, то вот краткий пересказ того, как происходит применение нового значения пункта pool_size :
У других пунктов настроек схема может быть попроще, но принципиально отличаться не будет.
Заключение
Если вы дочитали, а не пролистали, статью до этого места, вам представленный код мог показаться чересчур сложным для решаемой им задачи (а если раньше вы уже делали что-то похожее, возможно, наоборот, слишком простым). Нужно учитывать, что такое навороченное решение с почти реализованной транзакционностью не появляется за один присест. Это результат неоднократного рефакторинга, причём каждая следующая итерация по сути добавляла новый уровень ненаивности в реализацию. И я не знаю, сколько их ещё впереди. Наивные решения задач — это хорошо и эффективно в большинстве случаев, если только речь не про многопоточность. А многопоточность — это сложно.
Как просто написать распределенный веб-сервис на Python + AMQP
Привет, Хабр. Я уже довольно давно пишу на Python. Недавно пришлось разбираться с RabbitMQ. Мне понравилось. Потому что он без всяких проблем (понятно, что с некоторыми тонкостями) собирается в кластер. Тут я подумал: а неплохо бы его использовать в качестве очереди сообщений в кусочке API проекта, над которым я работаю. Сам API написан на tornado, основная мысль была в исключении блокирующего кода из API. Все синхронные операции выполнялись в пуле тредов.
Первое, что я решил, это сделать отдельный процесс(ы) «worker», который бы брал на себя всю синхронную работу. Задумал, чтобы «worker» был максимально прост, и делал задачи из очереди одну за другой. Скажем, выбрал из базы что-нибудь, ответил, взял на себя следующую задачу и так далее. Самих «worker»ов можно запустить много и тогда AMQP выступает уже в роли некоего подобия IPC.
Спустя некоторое время из этого вырос модуль, который берет на себя всю рутину связанную с AMQP и передачей сообщений туда и назад, а также сжимает их gzipом, если данных слишком много. Так родился crew. Собственно, используя его, мы с вами напишем простой API, который будет состоять из сервера на tornado и простых и незамысловатых «worker» процессов. Забегая вперед скажу, что весь код доступен на github, а то, о чем я буду рассказывать дальше, собрано в папке example.
Подготовка
Итак, давайте разберемся по порядку. Первое, что нам нужно будет сделать — это установить RabbitMQ. Как это делать я описывать не буду. Скажу лишь то, что на той-же убунте он ставится и работает из коробки. У меня на маке единственное, что пришлось сделать, это поставить LaunchRocket, который собрал все сервисы, что были установлены через homebrew и вывел в GUI:
Дальше создадим наш проект virtualenv и установим сам модуль через pip:
В зависимостях модуля умышленно не указан tornado, так как на хосте с workerом его может и не быть. А на веб-части обычно создают requirements.txt, где указаны все остальные зависимости.
Код я буду писать частями, чтобы не нарушать порядок повествования. То, что у нас получится в итоге, можно посмотреть тут.
Пишем код
Сам tornado сервер состоит из двух частей. В первой части мы определяем обработчики запросов handlers, а во второй запускается event-loop. Давайте напишем сервер и создадим наш первый метод api.
Благодаря coroutine в торнадо, код выглядит просто. Можно написать тоже самое без coroutine.
Наш сервер готов. Но если мы его запустим, и сходим на /, то не дождемся ответа, его некому обрабатывать.
Теперь напишем простой worker:
Итак, как видно в коде, есть простая функция, обернутая декоратором Task(«test»), где test — это уникальный идентификатор задачи. В вашем worker не может быть двух задач с одинаковыми идентификаторами. Конечно, правильно было бы назвать задачу «crew.example.test» (так обычно и называю в продакшн среде), но для нашего примера достаточно просто «test».
Сразу бросается в глаза context.settings.counter. Это некий контекст, который инициализируется в worker процессе при вызове функции run. Также в контексте уже есть context.headers — это заголовки ответа для отделения метаданных от ответа. В примере с callback-функцией именно этот словарь передается в _on_response.
Заголовки сбрасываются после каждого ответа, а вот context.settings — нет. Я использую context.settings для передачи в функции worker(ы) соединения с базой данных и вообще любого другого объекта.
Также worker обрабатывает ключи запуска, их не много:
URL подключения к базе и прочие переменные можно брать из переменный окружения. Поэтому worker в параметрах ждет только как ему соединиться c AMQP (хост и порт) и уровень логирования.
Итак, запускаем все и проверяем:
Работает, но что случилось за ширмой?
При запуске tornado-сервера tornado подключился к RabbitMQ, создал Exchange DLX и начал слушать очередь DLX. Это Dead-Letter-Exchange — специальная очередь, в которую попадают задачи, которые не взял ни один worker за определенный timeout. Также он создал очередь с уникальным идентификатором, куда будут поступать ответы от workerов.
После запуска worker создал по очереди на каждую обернутую декоратором Task очередь и подписался на них. При поступлении задачи воркер main-loop создает один поток, контролируя в главном потоке время исполнения задачи и выполняет обернутую функцию. После return из обернутой функции сериализует его и ставит в очередь ответов сервера.
После поступления запроса tornado-сервер cтавит задачу в соответствующую очередь, указывая при этом идентификатор своей уникальной очереди, в которую должен поступить ответ. Если ни один воркер не взял задачу, тогда RabbitMQ перенаправляет задачу в exchange DLX и tornado-сервер получает сообщение о том, что истек таймаут пребывания очереди, генерируя исключение.
Зависшая задача
Чтобы продемонстрировать, как работает механизм завершения задач, которые повисли в процессе выполнения, напишем еще один веб-метод и задачу в worker.
В файл master.py добавим:
И добавим его в список хендлеров:
Как видно из приведенного выше примера, задача уйдет в бесконечный цикл. Однако, если задача не выполнится за 3 секунды (считая время получения из очереди), main-loop в воркере пошлет потоку исключение SystemExit. И да, вам придется обработать его.
Контекст
Как уже упоминалось выше, контекст — это такой специальный объект, который импортируется и имеет несколько встроенных переменных.
Давайте сделаем простую статистику по ответам нашего worker.
В файл master.py добавим следующий handler:
Также зарегистрируем в списке обработчиков запросов:
Этот handler не очень отличается от предыдущих, просто возвращает значение, которое ему передал worker.
Теперь сама задача.
В файл worker.py добавим:
Функция возвращает строку, с информацией о количестве задач, обработанных workerом.
PubSub и Long polling
Теперь реализуем пару обработчиков. Один при запросе будет просто висеть и ждать, а второй будет принимать POST данные. После передачи последних первый будет их отдавать.
Напишем задачу publish.
Если же вам не нужно передавать управление в worker, можно просто публиковать прямо из tornado-сервера
Параллельное выполнение заданий
Часто бывает ситуация, когда мы можем выполнить несколько заданий параллельно. В crew есть для этого небольшой синтаксический сахар:
В этом случае, задаче будут поставлены две задачи параллельно и выход из with будет произведен по окончании последней.
Но нужно быть осторожным, так как какая-то задача может вызвать исключение. Оно будет приравнено непосредственно переменной. Таким образом, вам нужно проверить, не является ли test_result и stat_result экземплярами класса Exception.
Планы на будущее
Когда eigrad предложил написать прослойку, которой можно запустить любое wsgi приложение с помощью crew, мне эта идея сразу понравилась. Только представьте, запросы хлынут не на ваше wsgi приложение, а будут равномерно поступать через очередь на wsgi-worker.
Я никогда не писал wsgi сервер и даже не знаю, с чего начать. Но вы можете мне помочь, pull-requestы я принимаю.
Также думаю дописать client для еще одного популярного асинхронного фреймворка, для twisted. Но пока разбираюсь с ним, да и свободного времени не хватает.
Благодарности
Спасибо разработчикам RabbitMQ и AMQP. Замечательные идеи.
Также спасибо вам, читатели. Надеюсь, что вы не зря потратили время.
Эффективная многопоточность в Python
Хочу поделиться простым рецептом, как можно эффективно выполнять большое число http-запросов и других задач ввода-вывода из обычного Питона. Самое правильное, что можно было бы сделать — использовать асинхронные фреймворки вроде Торнадо или gevent. Но иногда этот вариант не подходит, потому что встроить event loop в уже существующий проект проблематично.
В моем случае уже существовало Django-приложение, из которого примерно раз в месяц нужно было выгрузить немного очень мелких файлов на AWS s3. Шло время, количество файлов стало приближаться к 50 тысячам, и выгружать их по очереди стало утомительным. Как известно, s3 не поддерживает множественное обновление за один PUT-запрос, а установленная опытным путем максимальная скорость запросов с сервера ec2 в том же датацентре не превышает 17 в секунду (что очень не мало, кстати). Таким образом, время обновления для 50 тысяч файлов стало приближаться к одному часу.
Питонисты с детства знают, что от использования потоков (тредов операционной системы) нет никакого толка из-за глобального лока интерпретатора. Но немногие догадываются, что как и любой лок, этот время от времени освобождается. В частности, это происходит при операциях ввода-вывода, в том числе и сетевых. А значит, потоки можно использовать для распараллеливания http-запросов — пока один поток ожидает ответа, другой спокойно обрабатывает результат предыдущего или готовит следующий.
Здесь concurrency — число рабочих потоков, upload — функция, выполняющую саму задачу, queryset — итератор объектов, которые по одному будут передаваться в задачу. Уже этот код при concurrency в 150 смог пропихнуть на сервера Амазона ≈450 запросов в секунду.
Тут необходимо замечание относительно задач: они должны быть потокобезопасны. Т.е. несколько паралельно выполняющихся задач не должны иметь общих ресурсов, либо должны ими правильно управлять. Глобальный лок интерпретатора тут плохой помощник — он не гарантирует, что выполнение потока не прервется в самом неподходящем месте. Если вы пользуетесь только urllib3, requests или boto, волноваться не о чем, они уже потокобезопасны. Про другие библиотеки нужно уточнять. Также потоконебезопасным может оказаться ваш собственный код.
Шло время, количество файлов стало приближаться к 200 тысячам. Как думаете, сколько памяти могут занимать 200 тысяч Django-моделей? А 200 тысяч фьючерсов? А 200 тысяч поставленных задач? Все вместе около гигабайта. Стало понятно, что посылать в экзекутор все сразу — не выход. Но почему бы не добавлять новые задачи по окончании предыдущих? В самом начале добавляем количество задач, равное количеству потоков, ведем учет сколько задач поставлено, сколько выполнено. Сами фьючерсы не храним, наружу не отдаем. Получается очень классная функция, которую можно использовать повторно (осторожно, это не окончательный вариант) :
Правда, есть проблема: стоит увеличить кол-во потоков, как начинают сыпаться исключения «ValueError: generator already executing». Код использует один и тот же генератор из всех потоков, поэтому рано или поздно два потока пытаются выбрать значения одновременно (на самом деле это может произойти когда потоков всего два, но с меньшей вероятностью). Это же касается и счетчиков, рано или поздно два процесса одновременно считают одно значение, потом оба прибавят единицу и оба запишут «исходное число + 1», а не «исходное число + 2». Поэтому всю работу с разделяемыми объектами нужно обернуть в локи.
Есть и другие проблемы. Нет обработки ошибок, которые могут произойти во время выполнения задачи. Если прервать выполнение с помощью ctrl+c, в основном потоке будет выброшено исключение, а остальные продолжат выполнение до самого конца, поэтому нужен механизм принудительного завершения очереди. У экзекутора как раз есть метод shutdown для этих целей и можно было бы отдавать экзекутор наружу, чтобы останавливать его, когда пользователь нажимает ctrl+c. Но есть вариант получше: можно создать фьючерс, который будет резолвится по окончании всех работ и подчищать экзекутор, если кто-то извне его отменит. Вот версия, в которой учтены все эти ошибки:
Больше не нужно тупо засыпать каждые 200 миллисекунд, можно засыпать по умному, ожидая завершения очереди. А в случае прерывания останавливать очередь.
Смеркалось. Шло время, количество файлов стало приближаться к 1,5 миллионам. Несмотря на то, что все выглядело так, как будто все работает с фиксированным потреблением памяти (кол-во тредов, фьючерсов и Django-моделей на протяжении всего выполнения не должно меняться), потребление памяти все равно росло. Оказалось, что queryset.iterator() работает немного не так, как ожидалось. Объекты действительно создаются только тогда, когда явно выбираются из итератора, а вот сырой ответ базы данных все равно выгребается драйвером сразу. Получается около 500 мегабайт на миллион строк. Решение этой проблемы довольно очевидно: нужно делать запросы не на все объекты сразу, а разделять порции. При этом следует избегать выборки со смещением, потому что запрос вида LIMIT 100 OFFSET 200000 на самом деле означает, что СУБД нужно пробежаться по 200100 записям. Вместо смещения следует использовать выборку по полю с индексом.
Здесь pk — скорее pagination key, нежели primary. Впрочем, зачастую primary хорошо подходит на эту роль. Такой итератор действительно расходует фиксированное количество памяти и работает не медленнее выборки за один раз. Но если увеличить кол-во потоков, возникает еще одна проблема. В Джанге соединения с базой данных являются локальными для потоков, поэтому, когда очередной поток делает запрос, создается новое соединение. Рано или поздно количество соединений доходит до критического числа и возникает исключение, подобное этому:
Правильным решением было бы использовать для всех потоков одно и то же соединение, т.к. мы уже ограничили возможность одновременно делать запросы из разных потоков. Стандартных средств для этого в Джанге нет, но это можно сделать с помощью хака, заменив объект threading.local на обычный объект:
Но надо понимать, что это убъет потокобезопасность базы данных во всем остальном приложении, поэтому такой вариант годится только для команд, запускаемых из консоли. Более гуманный вариант — закрывать соединение после каждого запроса, или после каждого элемента, что дает не сильно большой оверхэд.
Есть и третье решение: использовать отдельный поток, который будет общаться с базой данных, отдавая объекты в остальные потоки. Этот вариант ничего не ломает в остальном приложении и не привносит накладных расходов на постоянное переоткрытие соединений. Но его реализация довольно сложна и тянет не меньше чем на отдельную статью.