Глава 4. 20 библиотек Asyncio, которые вы не применяете (Но... Да ладно, не берите в голову)

поддерживал постоянные соединения со множеством клиентов,

получал сообщения от клиентов с каким- то целевым "названием канала",

доставлял эти сообщения всем прочим клиентам, подписанным на канал с тем же самым названием.

Каналы с префиксом /topic в названии, например, /topic/customer/registration управляются при помощи шаблона Издатель-подписчик (все подписчики канала получают все сообщения)

Каналы именуемые с префиксом /queue обрабатываются в соответствии с моделью точка-точка {P2P, одноранговой}, при которой сообщения в канале распространяются между подписчиками канала неким карусельным манером: каждый подписчик получает какое- то уникальное сообщение.

⁽¹⁾ Получаем самые первые четыре байта. Это префикс размера.

⁽²⁾ Эти четыре байта необходимо преобразовать в целое значение.

⁽³⁾ Теперь, когда мы знаем размер полезной нагрузки, считываем её из имеющегося потока.

⁽⁴⁾ Запись обратна к чтению: вначале мы отправляем значение длины своих данных, кодируя их в 4 байтах.

⁽⁵⁾ Затем отправляем собственно данные.

⁽⁶⁾ drain() гарантирует то что необходимые данные отправлены полностью. Без drain() какие- то данные всё ещё могут дожидаться своей очереди в имеющемся буфере отправки в то время как выполнен выход из данной сопрограммы.

⁽⁷⁾ Это здесь не существенно, но я снабдил функцию под запуск некоего сервера TCP. Данная последовательность останова уже обсуждалась ранее в нашем предыдущем разделе и я включаю это здесь исключительно для экономии места в последующих примерах кода. Останов сервера начинается через SIGINT или Ctrl-C.

⁽¹⁾ Импорт из нашего модуля msgproto.py.

⁽²⁾ Некая общая коллекция активных в настоящее время подписчиков. При каждом подключении какого- то клиента, тот обязан вначале отправить некое название канала, на который он подписывается. Очередь с двусторонним доступом (deque) удержит всех имеющихся подписчиков в каком- то определённом канале.

⁽³⁾ Данная функция сопрограммы client() произведёт какую- то долгоживущую сопрограмму для каждого нового соединения. Представляйте себе её как некий обратный вызов того сервера TCP, который запускается в run_server(). В этой строке я показал как можно получить адрес хоста и порт удалённого однорангового партнёра, например, для регистрации.

⁽⁴⁾ Наш протокол для клиентов состоит в следующем:

• При первом подключении некий клиент обязан отправить какое- то сообщение, содержащее тот канал, на который он подписывается (в данном случае, subscribe_chan).

• После этого, для обеспечения собственно жизни данного соединения клиент отправляет какое- то сообщение в канал вначале отправляя какое- то сообщение, содержащее имя своего целевого канала, за которым следует сообщение с данными. Наш брокер отправит такие сообщения с данными каждому клиенту, который подписан на данный канал.

⁽⁵⁾ Добавляем необходимый экземпляр StreamWriter в имеющуюся общую коллекцию подписчиков.

⁽⁶⁾ В бесконечном цикле ожидаем данные от этого клиента. Самое первое сообщение от какого- то клиента должно быть названием целевого канала.

⁽⁷⁾ Далее переходим к реальным данным для распространения в конкретном канале.

⁽⁸⁾ Получаем определённую очередь с двусторонним доступом (deque) подписчиков в своём целевом канале.

⁽⁹⁾ Некая особая обработка если данное название канала начинается с магического слова "/queue": в таком случае мы отправляем свои данные только одному из всех подписчиков, а не всем им. Это можно применять для разделения работы между каким- то пакетом исполнителей, вместо того чтобы следовать обычной схеме уведомлений pub-sub, при которой все подписчики в некотором канале получают все сообщения.

⁽¹⁰⁾ Именно по этой причине мы используем двустороннюю очередь, а не список: вращение по нашей deque это то как мы отслеживаем своего следующего клиента в строке для распространения "/queue". Это кажется затратным, пока вы не поймёте, что однократное обращение по двусторонней очереди является операцией O(1).

⁽¹¹⁾ Целевым является всякий идущий первым клиент; это изменяется после каждого кругового обхода.

⁽¹²⁾ Создаём некий список сопрограмм для отправки данного сообщения каждому записывающему, а затем распаковываем его в gather() с тем, чтобы мы могли дождаться завершения всех этих отсылок.

Замечание

Эта строка является негодной брешью в нашей программе, но может быть не очевидно почему: хотя может быть и правильным дождаться отправки всем подписчикам, которая происходит одновременно, что произойдёт если у нас имеется очень медленный клиент? В таком случае наш gather() завершится только когда самый тормозной подписчик получит свои данные. Мы не можем получать никакие иные данные от отправляющих их клиентов пока не завершатся все эти send_msg(). Это замедляет распространение всех сообщений до значения скорости самого медленного подписчика.

⁽¹³⁾ Когда мы покидаем сопрограмму client(), не забудьте удалить себя из имеющейся общей коллекции SUBSCRIBERS. К сожалению, это операция с O(n), которая может быть затратной при больших значениях n. Некая иная структура исправит это, но в данный момент мы утешаем себя пониманием того что соединения рассматриваются как обладающие большим временем жизни, следовательно события отключения единичны; а n вряд ли будет слишком большим (скажем ~ 10 000 в качестве грубой оценки); и в таком коде достаточно просто разобраться!

⁽¹⁾ Наш стандартный модуль uuid является удобным способом создания какй- то "уникальности" для данного ожидающего. Если вы при помощи этого вы запускаете множество экземпляров, каждый из них будет иметь свй собственный идентификатор и у вас будет возможность отслеживать в соответствующих журналах что происходит.

⁽²⁾ Открываем соединение со своим сервером.

⁽³⁾ Название канала для подписки является неким входным параметром, которые выбирается из args.listen. Перед отправки кодируем его в bytes.

⁽⁴⁾ По правилам нашего протокола (которые описаны в анализе кода нашего брокера ранее), самое первое, что мы должны сделать сразу после подключения, так это отправит то название канала, на который требуется подписка.

⁽⁵⁾ Этот цикл ничего не делает кроме того что ждёт появления данных в своём сокете.

⁽⁶⁾ Параметры командной строки для данной программы чтобы сделать более простым указание хоста, порта и названия канала, в котором следует выполнять ожидание.

⁽¹⁾ Как и для стороны ожидания, требуется некая идентичность.

⁽²⁾ Как и для стороны ожидания, добиваемся выполнения подключения.

⁽³⁾ В соответствии с правилами нашего протокола, самый первый момент, который следует выполнить сразу после подключения, состоит в получении самого названия канала для подписки; однако, так как мы являемся стороной отправки, мы на самом деле не озабочены подпиской на какие- либо каналы; тем не менее, так как наш протокол того требует, поэтому просто предоставляем некий нулевой канал для подписки (на самом деле нам не требуется ничего ожидать).

⁽⁴⁾ Отправляем значение канала для подписки.

⁽⁵⁾ Получаемый из командной строки параметр args.channel предоставляет тот канал, в который мы желаем отправлять сообщения. Отметим, что его следует преобразовать в байты перед отправкой.

⁽⁶⁾ Применение itertools.count() похоже на цикл while True, за исключением того что вы получаете для использования некую переменную итераций. Мы применяем это при отладке обмена сообщениями, так как таким образом проще отслеживать какое сообщение откуда отправлено.

⁽⁷⁾ Значение задержки между отправляемыми сообщениями является неким получаемым на входе параметром, args.interval. Наша следующая строка вырабатывает полезную нагрузку данного сообщения. Это либо строка байт предписанного размера (args.size), либо некое описательное сообщение. Такая гибкость служит исключительно для проверки.

⁽⁸⁾ Отправляем! Заметим, что здесь присутствуют два сообщения: самым первым является название целевого канала, а второе представляет собственно полезную нагрузку.

⁽⁹⁾ Как и в случае стороны ожидания, имеется некий пакет параметров командной строки для регулировок нашего отправителя: "channel" определяет целевой канал для отправки, в то время как "interval" управляет значение задержки между отправками. Значение параметра "size" управляет размером полезной нагрузки каждого сообщения.

⁽¹⁾ В нашей предыдущей реализации имелся только один SUBSCRIBERS; теперь в глобальном наборе присуствуют SEND_QUEUES и CHAN_QUEUES. Именно это является результатом полного разделения приёма и отправления данных. SEND_QUEUES имеет одну запись очереди для каждого соединения с клиентом: все подлежащие отправке этому клиенту данные должны помещаться в эту очередь. (Если вы быстро заглянете вперёд, сопрограмма send_client() выгрузит SEND_QUEUES и отправит её.)

⁽²⁾ До текущего момента в нашей функции сопрограммы client() весь код был в точности тем же, что и в нашем образце сервера: принималось название канала подписчика и мы добавляли соответствующий экземпляр StreamWriter для такого нового клиента в свою глобальную коллекцию SUBSCRIBERS.

⁽³⁾ А вот это новинка: мы создаём некую задачу с долгим временем жизни которая будет выполнять всю отправку данных такому клиенту. Эта задача будет исполняться независимо в виде какой- то отдельной сопрограммы и будет выгружать для отправки сообщения поддерживаемой ею очереди, SEND_QUEUES[writer]

⁽⁴⁾ Теперь мы находимся внутри того цикла, в котором мы принимаем данные. Помните, что мы всегда получаем два сообщения: одно для своего названия целевого канала, а второе собственно для данных. Мы намерены создать каую- то новую, выделенную Queue для каждого целевого канала, и именно этому служит CHAN_QUEUES: всякий раз, когда клиент желает поместить данные в какой- то канал, мы обираемся размещать такие данные в соответствующей очереди, а затем немедленно возвращаемся к ожиданию дополнительных данных. Такой подход отсоединяет имеющееся распределение обмена сообщениями от необходимого приёма сообщений этим же клиентом.

⁽⁵⁾ Если уже имеется некая очередь для данного целевого канала, пользуемся ею.

⁽⁶⁾ Для этого канала создаём некую выделенную задачу с длительным временем жизни. Данная сопрограмма, chan_sender(), будет отвечать за забор данных из очереди этого канала и распространения этих данных подписчикам.

⁽⁷⁾ Помещаем свои вновь полученные данные в очередь некого специфического канала. Отметим, что если данная очередь заполнится, мы будем выполнять здесь ожидание пока не освободится место для наших новых данных. В процессе ожидания в этом месте мы не будем считывать никакие новые данные из своего сокета, что означает, что нашему клиенту придётся со своей стороны выполнять ожидание отправки новых данных в соответствующий сокет. Это не обязательно плохой момент, так как он сообщает данному клиенту о так называемом противодавлении - back­ pressure. (В противном случае вы можете здесь выбрать сброс сообщений, если это подходит для вашего варианта применения.)

⁽⁸⁾ После того как это соединение закрывается, самое время для его очистки! Та долгоиграющая задача, которую мы создавали для отправки данных этому клиенту, send_task, может быть остановлена помещением None в её очередь, SEND_QUEUES[writer] (проверьте соответствующий код для send_client()). Существенно применять какое- то значение в данной очереди вместо того чтобы полностью прекращать её, так как в этой очереди уже могут иметься данные и мы хотим отправки этих данных прежде чем завершить send_client().

⁽⁹⁾ Ожидаем завершения данной задачи отправителя.

⁽¹⁰⁾ Удаляем соответствующую запись в своей коллекции SEND_QUEUES (а в своей следующей строке мы также удаляем соответствующий sock из своей коллекции SUBSCRIBERS, как и ранее).

⁽¹¹⁾ Наша функция сопрограммы send_client() очень похожа на пример из текста данной книги по извлечению исполнителя из очереди. Отметим как данная сопрограмма выполнит выход если поместить None в соответствующую очередь. Также укажем на то как мы гасим CancelledError внутри данного цикла: именно поэтому мы желаем чтобы данная задача закрывалась исключительно по получению None в своей очереди. Таким образом все придерживаемые в этой очереди данные могут быть отправлены до её останова.

⁽¹²⁾ chan_sender() является некоторой необходимой логикой распространения для какого- то канала: она отправляет данные из какого- то выделенного экземпляра Queue канала всем своим подписчикам в этом канале. Но что произойдёт если пока ещё нет никаких подписчиков для этого канала? Мы всего лишь немного подождём и повторим попытку вновь (Заметим, что соответствующая очередь этого канала, то есть CHAN_QUEUES[name], тем не менее продолжит заполняться.)

⁽¹³⁾ Как и ранее в своей предыдущей реализации брокера, мы делаем нечто особенное для каналов, название которых начинается с "/deque": мы вращаем свою deque (двустороннюю очередь, колоду) и отправляем только самую первую запись. Это действует как некая недоделанная система балансировки нагрузки, так как каждый подписчик получает различные сообщения из той же самой очереди. Для всех прочих каналов все подписчики получают все имеющиеся сообщения.

⁽¹⁴⁾ Здесь мы выполняем ожидание данных этой очереди. В своей следующей строке мы выполняем выход в случае получения None. В настоящее время это не является предметом переключения где бы то ни было (следовательно, данная сопрограмма chan_sender() живёт постоянно); однако если впоследствии была добавлена логика очистки задач этого канала, скажем, в случае отсутствия активности на протяжении какого- то времени, именно здесь это следует делать.

⁽¹⁵⁾ Данные были получены, поэтому настало время их отправки подписчикам. Отметим что мы здесь не выполняем отправку: мы помещаем соответствующие данные в собственную очередь отправки каждого подписчика. Дакое разъединение необходимо чтобы гарантировать то, что какой- то медленный подписчик не будет замедлять получение данных кому- то ещё. И более того, если соответствующий подписчик настолько медленный, что его очередь отправки переполняется, мы не помещаем эти данные в его очередь, т.е. они теряются.

⁽¹⁾ Первоначально Twisted требовал создания экземпляров Deferred, а также добавления обратных вызовов к такому экземпляру в качестве своего метода построения асинхронных программ. Несколько лет назад был добавлен соответствующий декоратор @inlineCallbacks, который видоизменил генераторы в качестве сопрограмм.

⁽²⁾ Хотя @inlineCallbacks позволил нам писать код, который был линейным при своём возникновении (в отличии от обратных вызовов), всё же требовался некий хак, например, вызов defer.returnValue(), который был именно тем способом, которым возвращались значения из сопрограмм @inlineCallbacks.

⁽³⁾ Здесь мы можем обнаружить yield (оператор производства), который делает эту функцию неким генератором. Для того чтобы работал @inlineCallbacks, должен иметься по крайней мере один yield для осуществления декорации данной функции.

⁽¹⁾ Именно таким образом вы сообщаете Twisted о необходимости применять соответствующий цикл событий asyncio в качестве его основного reactor. Отметим, что данная строка обязана предшествовать самому импорту reactor из twisted.internet в нашей следующей строке.

⁽²⁾ Всякий знакомый с программированием Twisted распознает эти импорты. У нас нет места пояснять их здесь, но в своей сердцевине наш reactor является соответствующей версией Twisted необходимого цикла asyncio, а defer и task являются пространствами имён для инструментов по работе с составлением расписаний сопрограмм.

⁽³⁾ Увидеть async def здесь, в некоторой программе Twistes, выглядит как ужасно неуместное, но это именно то, что в конечном итоге даёт нам наша новая поддержка для async/await: возможность применять естественные сопрограммы непосредственно в программах Twisted!

⁽⁴⁾ В нашм более древнем мире @inlineCallbacks, вам пришлось бы применять здесь yield, но теперь мы можем тут воспользоваться await, точно также как и в коде asyncio. Оставшаяся часть данной строки, deferLater, является неким альтернативным способом того же самого, чем является asyncio.sleep(1). Мы выполняем await некоего фьючерса, в котором через 1 секунду будет зажжён некий ничего не выполняющий обратный вызов.

⁽⁵⁾ ensureDeferred() выступает в роли версии Twisted планирования какой- то сопрограммы. Это аналогично loop.create_task() или asyncio.ensure_future().

⁽⁶⁾ Запуск необходимого reactor в точности тот же что и для loop.run_forever() в asyncio.

queue.Queue: некая очередь с ",блокировкой", обычно применяемая для взаимодействия и буферизации между потоками (threads)

asyncio.Queue: некая совместимая с asyncio очередь, обычно используемая при взаимодействии и буферизации между сопрограммами (coroutines)

⁽¹⁾ Создаём некую очередь Janus. Отметим, что точно так же как и в случае с asyncio.Queue, очередь Janus будет связана с неким определённым циклом событий. Как обычно, если вы не предоставляете предписанный параметр loop, внутренним образом будет применяться соответствующий стандартный вызов get_event_loop().

⁽²⁾ Наша функция сопрограммы main() просто выполняет ожидание данных в какой- то очереди. Данная строка подвешивается до тех пор, пока нет данных, в точности как и для asyncio.Queue. Данный объект очереди имеет два "лица" {в точном соответствии с названием}: это именуется async_q, и оно предоставляет API очереди, совместимый с асинхронными приложениями.

⁽³⁾ Выводим некое сообщение.

⁽⁴⁾ Внутри нашей функции data_source() вырабатывается некое случайное целое значение, которое применяется как некая продолжительность сна, а также как какое- то значение данных. Отметим, что вызов time.sleep() является блокирующим, поэтому данная функция должна исполняться в каком- то потоке.

⁽⁵⁾ Помещаем полученные данные в имеющуюся очередь Janus. Это отображает второе "лицо" нашей очереди Janus: sync_q, которое предоставляет наш стандартный API Queue с блокировкой.

⁽¹⁾ Создан некий экземпляр Application.

⁽²⁾ Создан какой- то корень, причём соответствующая целевая сопрограмма hello() определяется как соответствующий обработчик.

⁽³⁾ Запускается веб приложение.

⁽¹⁾ Наша функция news() является соответствующим обработчиком для URL /news в нашем сервере. Она возвращает окончательную страницу HTML, отображающую все имеющиеся заголовки.

⁽²⁾ Здесь у нас имеется толлько два новостных вебсайта для выдёргивания: CNN и Al Jazeera. Легко можно добавить дополнительные, но затем также потребуется добавить дополнительные обработчики данных, подобные представленным функциям cnn_articles() и aljazeera_articles(), каждая из которых имеет индивидуальные настройки извлечения данных заголовков.

⁽³⁾ Для каждого новостного сайта мы создаём некую задачу для извлечения и обработки данных соответствующей страницы HTML для их передовиц. Отметим, что мы распаковываем соответствующие кортежи ((*s)), так как наша функция сопрограммы news_fetch() получает оба URL и функцию обработки данных в качестве параметров. Каждый news_fetch() будет возвращать перечень кортежей в виде (<article URL>, <article title>).

⁽⁴⁾ Все задачи собираются воедино в отдельный Future (gather() возвращает некий фьючерс, представляющий значение состояния всех подлежащих выработке задач), а затем мы немедленно выполняем await относительно завершения такого фьючерса. Данная строка будет в подвешенном состоянии пока не завершится соответствующий фьючерс.

⁽⁵⁾ Так как все имеющиеся задачи news_fetch() теперь завершены, мы собираем все полученные результаты в некий словарь. Отметим как вложенные содержания понятий применяются для выполнения итераций по задачам, а затем и по списку кортежей, возвращаемых каждой задачей. Мы также используем f- строки для непосредственной подстановки данных, в том числе и сам "вид" страницы, что будет применяться в CSSS для обозначения цветом фона соответствующего div.

⁽⁶⁾ В этом словаре значением ключа является сам заголовок, а его значением некая строка HTML для какого- то div, которая будет отображаться в нашей странице результата.

⁽⁷⁾ Наш веб сервер собирается вернуть HTML. Мы выгружаем данные HTML из некоторого локального файла с названием index.html, этот файл предоставляется в наших Дополнениях, если вы пожелаете восстановить данный пример самостоятельно.

⁽⁸⁾ Мы подставляем собранные DIV заголовков в общий шаблон и возвращаем полученную страницу в браузер своего клиента. Созданная страница отображена на Рисунке 4-1.

⁽⁹⁾ Здесь, внутри соответствующей функции сопрограммы news_fetch(), у нас имеется крошечный шаблон для попадания в свой API Splash (в моём случае он исполняется в некотором локальном контейнере docker с портом 8050). Здесь мы демонстрируем как aiohttp может применяться в качестве некоторого клиента HTPP.

⁽¹⁰⁾ Самый стандартный путь состоит в создании некоего экземпляра ClientSession() и последующем применении метода get() в имеющемся экземпляре сеанса для выполнения необходимого вызова REST. В нашей следующей строке получаются данные отклика. Отметим, что поскольку мы всегда оперируем в сопрограммах, причём применяя async with и await, эти сопрограммы никогда не выполняют блокировку: у нас будет иметься возможность обрабатывать многие тысячи таких запросов, даже хотя эти операции, т.е. news_fetch(), могут быть относительно медленными, поскольку мы выполняем веб вызовы внутренним образом.

⁽¹¹⁾ После получения необходимых данных вызываем требуемую функцию обработки данных. Возвращаясь назад, вспоминаем, что для CNN это будет cnn_articles(), а для Al Jazeera aljazeera_articles().

⁽¹²⁾ У нас очень мало места для краткого обзора обработки данных. После получения данных соответствующей страницы мы применяем библиотеку Beautiful Soup 4 для выделения заголовков.

⁽¹³⁾ Наша функция match() возвратит все имеющие соответствие теги (я вручную проверил соответствующие исходные коды HTML исходные коды этих новостных вебсайтов чтобы вывести какие комбинации фильтров выделяют наилучшие теги), а затем мы возвращаем какой- то список кортежей, соответствующий установленному формату (<article URL>, <article title>).

⁽¹⁴⁾ Это аналогично обработке данных для Al Jazeera. Соответствующее условие match() слегка другое, но иным способом выдаёт то же самое из CNN.

⁽¹⁾ Сокеты ØMQ имеют типы! Это некий сокет PULL. Вы можете представлять себе его как некий вид сокета "доступного только на получение", который будет запитываться неким иным сокетом "только для отправки", который будетсокетом типа PUSH.

⁽²⁾ Данный сокет с типом SUB является иным видом сокета "доступного только на получение", и он будет питаться сокетом типа PUB, который служит "только для отправки".

⁽³⁾ Если вам необходимо перемещать данные между множеством сокетов в каком- то приложении ØMQ потока, вам понадобится некое средство опроса (poller). Это происходит по причине того, что такие сокеты не сберегают поток, поэтому вы не можете выполнять recv() в различных сокетах из разных потоков [4: На самом деле вы можете делать это пока ваши подлежащие использованию в различных потоках сокеты создаются, применяются и целиком уничтожаются в своих собственных потоках. Это возможно, но трудно сделать, причём многие люди пытаются добиться этого. Именно по этой причине настолько сильна рекомендация применять единый поток и механизм опроса.]

⁽⁴⁾ Это работает аналогично системному вызову select(). Данная система опроса не будет блокирована когда имеются данные готовыми на приём в одном из своих зарегистрированных сокетов и затем она поднимается для вас чтобы выгрузить эти данные и сделать с ними что- то. Этот большой блок if является тем как вам приходится обнаруживать правильный сокет.

⁽¹⁾ Этот образец кода делает то же самое что и ранее, за исключением того, что теперь мы пользуемся преимуществами сопрограмм чтобы поменять всё. Теперь мы можем иметь дело с каждым сокетом по отдельности. Мы создали две функции сопрограмм, одну для каждого сокета, а вторую для имеющегося сокета PULL.

⁽²⁾ В pyzmq мы пользуемся поддержкой asyncio, что означает, что все send() и recv() обязаны применять соответствующее ключевое слово await. Наш Poller больше не появляется где бы то ни было, так как он интегрирован в сам цикл событий asyncio.

⁽³⁾ Именно это обработчик для нашего сокета SUB. Его структура очень похожа на соответствующий обработчик сокета PULL, но тот не является предметом в этом случае. Если бы потребовалась более сложная логика, мы могли бы запросто добавить его тут, причём полностью заключив только внутри кода обработчика данного SUB.

⁽⁴⁾ И вновь: совместимые с asyncio сокеты требуют такого ключевого слова await при отправке и приёме.

⁽⁵⁾ Для запуска необходимого цикла событий asyncio и создания задач для каждого сокета требуются эти дополнительные строки. Я срезал здесь несколько углов и опустил всю обработку ошибо и очистку, так как хочу выделить само воздействие на компоновку кода в дальнейшем.

Уровень приложения

Этот уровень содержит все наши приложения. Примерами могут служить "индивидуальные" микрослужбы, микрослужбы "резервирования", некая микруслужба "электронной почты" и тому подобное. Мы добавляем некий "передающий" сокет ØMQ в каждое из своих приложений. Этот сокет будет отправлять измерения производительности в какой- то центральный сервер.

Уровень сбора информации

Наш центральный сервер выставит какой- то сокет ØMQ для сбора необходимых данных от всех запущенных экземпляров приложений. Этот сервер также будет обслуживать некуб веб страницу для показа графиков производительности во времени, а также наш сервер будет в реальном времени выдавать потоком те данные, которые приходят к нему!

Уровень виртуализации

Это сама веб страница подлежащая обслуживанию. Мы будем отображать все собираемые данные в каком- то наборе графиков и эти диаграммы будут в живую обновляться в реальном масштабе времени. Для упрощения примеров нашего кода мы воспользуемся библиотекой JavaScript Smoothie Charts, которая предоставляет все необходимые функции стороны клиента.

⁽¹⁾ В версиях pyzmq, предшествующих 17.0.0, требовалось применять в явном виде данную команду zmq.asyncio.install() чтобы разрешить поддержку Asyncio. На момент написания версия 17 находилась в бета версии {Прим. пер.: однако на момент перевода всё уже хорошо.}

⁽²⁾ Эта функция сопрограммы запустит некую сопрограмму с долгим временем жизни, последовательно отправляющую данные в свой процесс сервера.

⁽³⁾ Создаём некий сокет ØMQ! Имеются различные разновидности сокетов. Именно этот имеет тип PUB, и он допускает отправку сообщений в одном направлении в другой сокет ØMQ. Этот сокет имеет - как об этом гласит руководство ØMQ - суперсилу. Он будет автоматически обрабатывать все повторные подключения и буферировать логику для вас.

⁽⁴⁾ Подключаемся к своему серверу.

⁽⁵⁾ Наша последовательность останова управляется KeyboardInterrupt, впоследствии останавливаясь. Когда принимается соответствующий сигнал мы прекращаем все свои задачи. Здесь мы обрабатываем соответствующий возбуждённую CancelledError при помощи удобного диспетчера контекста suppress() из модуля стандартной библиотеки contextlib.

⁽⁶⁾ Выполняем итерации на постоянной основе, отправляя данные в свой сервер.

⁽⁷⁾ Так как ØMQ знает как работать с целыми сообщениями, а не просто с обрезками какого- то потока данных, он открывает свою дверь для стопки полезных обёрток вокруг обычной идиомы sock.send(): в данном случае мы применяем один из таких вспомогательных методов, send_json(), который автоматически выстраивает все аргументы в JSON. Это позволяет нам напрямую применять dict().

⁽⁸⁾ Надёжный способ передачи информации даты и времени состоит в следовании формата ISO 8601. Это в особенности справедливо если вы обмениваетесь данными даты и времени между программным обеспечением, написанным на различных языках программирования, так как подавляющее большинство реализаций языков способно работать с данным стандартом.

⁽⁹⁾ Чтобы закончить здесь мы должны получить исключительную ситуацию CancelledError, приводящую в результате к прекращению задачи. Наш сокет ØMQ должен быть закрыт чтобы позволить останов программы.

⁽¹⁰⁾ Эта функция main() символизирует самое настоящее приложение микрослужбы. Данное суммирование произвольных чисел производит поддельную работу, просто чтобы выдавать нам некие ненулевые данные для просмотра на нашем уровне виртуализации слегка позднее.

⁽¹¹⁾ Мы собираемся создать множество экземпляров данного приложения, поэтому будет удобно иметь возможность различать их (позднее на диаграммах) при помощи некоего параметра ­­color.

⁽¹²⁾ При получении сигнала SIGINT (например, при нажатии Ctrl­C), планировщик вызывает останов данного цикла.

⁽¹³⁾ Создаём и собираем задачи для каждой из имеющихся функций сопрограммы.

⁽¹⁴⁾ Получив соответствующий сигнал останова прекращаем все задачи. Это возбуждает некую CancelledError внутри всех тех сопрограмм, которые представлены в нашей группе tasks. После завершения, всё ещё требуется запускать эти задачи для их окончания, предоставляя им возможность обработать надлежащим образом такое завершение. К примеру, мы обязаны закрыть имеющийся сокет ØMQ чтобы выполнить окончательное завершение.

⁽¹⁵⁾ Наконец может быть прекращён контекст соответствующего ØMQ.

⁽¹⁾ Одна половина этой программы будет получать данные от прочих приложений, а другая половина будет предоставлять данные браузерам клиентов через SSE (server­sent events , события отправки сервером). Мы применяем WeakSet() для отслеживания всех подключённых в настоящее время веб клиентов. Каждый подключённый клиент будет иметь некий связанный с ним экземпляр Queue(), поэтому такой идентификатор connections на самом деле некий набор очередей.

⁽²⁾ Повторно вызывая это на своём уровне приложения мы применяем некий сокет zmq.PUB; здесь, на уровне сбора информации мы пользуемся его партнёром, с соответствующим типом сокета zmq.PUB. Этот сокет ØMQ может выполнять только приём, и при этом не выполняет никакой отправки.

⁽³⁾ Для данного типа сокета zmq.SUB требуется предоставлять некое имя подписки, но для своих целей мы просто проверяем что всё идёт своим чередом, поэтому тут у нас пустой название темы.

⁽⁴⁾ Тут мы подвязываемся к своему сокету zmq.SUB. Задумайтесь на мгновение об этом! В конфигурациях "pubsub" вам обычно приходится делать сторону публикации своего сервера (bind()) и сторону подписки соответствующего клиента (connect()). ØMQ применяет иной подход: любая сторона может выступать в качестве такого сервера. Для нашего случая применения это важно, так как каждый из экземпляров нашего уровня приложения будет подключаться к одному и тому доменному имени сервера соединения и нет никакого иного обходного пути.

⁽⁵⁾ Имеющаяся в pyzmq поддержка asyncioпозволяет нам выполнять await данных со стороны наших подключённых прикладных программ. И не только это, так как все входящие данные будут автоматически распаковываться из JSON (да, это означает, что data представлены в виде dict()).

⁽⁶⁾ Повторный вызов этого нашего набора connections удержит какую- то очередь для каждого веб клиента? Теперь, когда данные были получены, самое время отправить их всем клиентам: эти данные помещаются в каждую из очередей.

⁽⁷⁾ Наша функция сопрограммы feed() создаст сопрограммы для каждого подключённого веб клиента. Для активной доставки данных соответствующему клиенту внутри применяются события отправки сервером.

⁽⁸⁾ Как уже описывалось ранее, каждый веб клиент будет иметь свой собственный экземпляр queue для приёма данных из соответствующей сопрограммы collector(). Такой экземпляр queue добавляется в имеющийся набор connections, но поскольку connections является неким неустойчивым набором, соответствующая запись будет автоматически удаляться из connections когда соответствующая очередь покидает сферу, т.е. когда отключается некий веб клиент. Неустойчивые ссылки (weakrefs) великолепны для упрощения таких видов задач резервирования.

⁽⁹⁾ Имеющийся пакет aiohttp_sse предоставляет необходимый диспетчер контекста sse_response(). Он предоставляет нам некую сферу внутри тех данных, для которых мы выполняем активную доставку своему веб клиенту.

⁽¹⁰⁾ Мы остаёмся подключёнными к соответствующему веб клиенту и ожидаем данные для очереди этого конкретного клиента.

⁽¹¹⁾ Как только поступают необходимые данные (внутри collector()), они отправляются необходимому подключённому веб клиенту. Отметим, что здесь мы повторно инициализируем соответствующий dict данных. Некая оптимизация в показанном здесь коде позволит избежать распаковки JSON в (collector() и вместо этого воспользоваться (sock.recv_string чтобы обойти потребность упорядочения в обо конца. Конечно, в каком- то реальном сценарии вы можете пожелать в любом случае выполнять распаковку в своём средстве сбора информации и выполнять некую проверку этих данных прежде чем отправлять их в браузер своего клиента. Так много возможностей!

⁽¹²⁾ Оконечная точка index() является самой первой загружаемой страницей и здесь мы обслуживаем некий файл со статистическими данными с названием charts.html.

⁽¹³⁾ Наша библиотека aiohttp предоставляет возможности для вас подхватывать дополнительные сопрограммы с длительным временем жизни, которые вам могут понадобиться. При помощи соответствующей сопрограммы collector() у нас имеется именно такое положение дел, поэтому мы создаём некую сопрограмму запуска start_collector() и какую- то сопрограмму останова. Они будут вызываться на протяжении определённых этапов последовательностей запуска и останова aiohttp. Отметим, что мы добавили необходимую задачу средства сбора информации в саму нашу app, которая реализует какой- то протокол установки соответствия с тем, чтобы вы могли применять его в виде dict.

⁽¹⁴⁾ Здесь вы можете увидеть что мы получаем свою сопрограмму collector() из соответствующего идентификатора app и вызываем в ней cancel().

⁽¹⁵⁾ Наконец, вы можете увидеть где прицепляются необходимые индивидуальные сопрограммы запуска и останова: соответствующий экземпляр app предоставляет особые точки входа дополнительного ПО, к которым в конец могут быть добавлены ваши персональные сопрограммы.

⁽¹⁾ Создаём некий новый экземпляр EventSource в соответствующем URL /feed. Соответствующий браузер подключится к /feed в нашем сервере, metric­server.py. Заметим, что этот браузер автоматически попытается выполнить повторное подключение в случае утраты такого подключения. События отправки сервером (SSE) часто остаются недооценёнными, но имеется множество случаев в которых такая простота SSE может быть предпочтительной для веб сокетов.

⁽²⁾ Это событие onmessage() зажигается всякий раз когда наш сервер отправляет данные. Здесь наши данные подвергаются синтаксическому разбору в качестве JSON.

⁽³⁾ Повторный вызов данного идентификатора cpu является некоторой установкой соответствия цвета экземпляру TimeSeries(). В данном случае мы получаем некую временную последовательность и помещаем в её конец данные. Мы также получаем соответствующий временной штамп и выполняем его разбор для получения правильного формата, необходимого нашей диаграмме.

⁽¹⁾ Я скрыл некоторые инструменты подготовки в небольшом модуле util для упрощения вещей и сохранении самых центральных сообщений.

⁽²⁾ Наш класс Database даёт нам некий диспетчер контекста, который создаст для нас некую новую базу данных - в данном случае с наименованием test - и уничтожит эту базу данных при покидании этого диспетчера контекста. Это становится очень полезным при экспериментах с идеями в коде. Так как никакое состояние не переносится между экспериментами, вы всякий раз начинаете с чистой базой данных. Отметим, что данный диспетчер контекста является async with; мы рассмотрим это подробнее впоследствии, но в данный момент основной областью нашего фокуса в данной демонстрации является то, что происходит внутри нашей сопрограммы demo().

⁽³⁾ Yfi lbcgtnxth rjyntrcnf Database снабжает нас неким экземпляром Connection, который незамедлительно применяется для создания новой таблицы, users.

⁽⁴⁾ Вставляем новую запись. Хотя мы и можем применять для выполнения данной вставки .execute(), основным преимуществом использования fetchval() является то, что мы можем получать соответствующий id вновь вставляемой записи, который будет сохраняться в нашем идентификаторе pk.

	Замечание
	Мы применяем параметры ($1 и $2) для проброса данных в свой запрос SQL. Никогда не пользуйтесь интерполяцией строк или конкатенацией при построении запросов, так как это создаёт риск для безопасности!

⁽⁵⁾ При извлечении данных, намного более полезным будет применять методы на основе fetch, так как они возвратят объекты Record. asyncpg автоматически приводит в соответствие большинство типов данных в надлежащие типы данных для Python.

⁽⁶⁾ Мы немедленно применяем вспомогательную функцию get_row() для отображения вновь вставленной записи.

⁽⁷⁾ Мы изменили данные воспользовавшись соответствующей командой UPDATE для SQL. Это достаточно незначительная модификация: значение года в дате рождения изменяется на один год. Как и ранее, это осуществляется при помощи метода соединения execute(). Оставшаяся часть данного кода демонстрирует далее те же самые структуры, которые мы видели ранее, а также DELETE, за которым следует другой print(), происходящий несколькими строками позднее.

⁽⁸⁾

⁽¹⁾ Наш класс Database всего лишь некий предполагаемый диспетчер контекста для создания и удаления какой- то базы данных из экземпляра PostgreSQL. Необходимое название базы данных передаётся в имеющийся конструктор.

⁽²⁾ (Примечание: Данная последовательность сносок в коде намеренно отличается от данного списка.) Данные диспетчер контекста является именно асинхронным. Вместо обычных методов __enter__() и __exit__() у нас имеются их заменители __aenter__() и __aexit__(). Здесь, на стороне входа, мы создадим необходимую базу данных и вернём подключение к этой новой базе данных.

⁽³⁾ server_command() является другим вспомогательным методом, определяемым несколькими строками позднее. Мы применяем его для запуска необходимых команд при создании своей новой базы данных.

⁽⁴⁾ Для вновь созданной базы данных делается некое соединение. Отметим, что я жёстко закодировал определённые подробности о данном подключении: это сделано умышленно, так как я стараюсь сохранять данный образец кода небольшим. Вы запросто можете сделать его универсальным введя такие поля как имя пользователя, название хоста и номер порта.

⁽⁵⁾ На стороне выхода из своего диспетчера контекста мы закрываем текущее подключение и ...

⁽⁶⁾ ... уничтожаем базу данных.

⁽⁷⁾ Для завершённости решения, именно этот метод утилиты служит для запуска команд в самом сервере PostgreSQL. Для данной цели он создаёт некое подключение, запускает все определённые команды и выполняет выход.

⁽⁸⁾ Это сюрприз и он будет гвоздём программы в нашем следующем примере!

Методом запроса HTTP (­X) является PUT, а не POST.

Наш URL теперь нуждается в соответствующем поле id.

⁽¹⁾ Всё вплоть до этой строки является установленным по умолчанию запуском регистрации сообщений sanic.

⁽²⁾ Как это уже пояснялось, самый первый GET имеет результатом промах кэшировнаия, так как наш сервер только что запущен.

⁽³⁾ Это получено из curl ­X GET. Я добавил некую функциональность определения времён в свои оконечные точки API. В данном случае соответствующая обработка данного запроса GET требует ~4 мс.

⁽⁴⁾ Второй GET возвращает данные из своего кэша, и намного быстрое время, ~100х, указывает что эти данные теперь возвращаются из кэша.

⁽¹⁾ Вспомогательная утилита Database описывалась ранее. Она предоставляет те методы, которые требуются для подключения к нашей базе данных.

⁽²⁾ Два дополнительных инструмента я на скорую руку смастерил воедино для регистрации значения прошедшего времени для каждого оконечного пункта API. Мы применяли его в предыдущем обсуждении при выявлении того, когда некий GET возвращался из своего кэша. Сами реализации для aelapsed() и aprofiler() не важны для данного примера, но вы можете получить их в наших Дополнениях.

⁽³⁾ Создан необходимый основной экземпляр прикладного приложения Sanic.

⁽⁴⁾ Данная функция сопрограммы служит для создания новых записей покровителей. В некотором вызове add_route(), далее вниз по нашему коду, new_patron() связывается с соответствующей оконечной точкой /patron, причём только для надлежащего метода POST. Наш декоратор @aelapsed не является частью основного API Sanic: он является моим собственным новшеством, в основном с целью регистрации значений времён для каждого вызова.

⁽⁵⁾ Sanic предоставляет немедленный разбор получаемых данных .json при помощи соответствующего атрибута в получаемом объекте request.

⁽⁶⁾ Наша модель model, которую мы импортировали ранее, является основной моделью для нашей таблицы покровителей в общей базе данных. Мы рассмотрим её подробнее в своём следующем листинге кода. А на текущий момент просто уясните, что все наши запросы и SQL к базе данных находятся в этом модуле model. Здесь мы пробрасываем соответствующий пул соединений для своей базы данных, и именно они являются одним и тем же шаблоном для всех наших операций с нашей моделью базы данных в данной функции и в нашем классе PatronAPI ниже.

⁽⁷⁾ Будет создан новый первичный ключ, id, и именно он возвращается обратно вызывающей стороне в виде JSON.

⁽⁸⁾ В то время как создание обрабатывается в нашей функции new_patron(), все прочие взаимодействия обрабатываются в данном представлении на основе класса, который удобно предоставляется Sanic. Все имеющиеся в данном классе методы ассоциируются с одним и тем же URL, /patron/<id:int>, который вы можете увидеть далее вниз по коду в соответствующей add_route() ближе к концу. Отметим, что необходимый параметр URL id будет передаваться во всякий вызываемый метод, причём этот параметр требуется для всех трёх оконечных точек.

Вы можете не опасаться проигнорировать соответствующий аргумент metaclass: всё что он выполняет, это обёртка вокруг каждого метода при помощи декоратора @aelapsed с тем чтобы значения времени выводились в нашем журнале регистраций. Это не является частью API Sanic, и опять- таки является моим собственным нововведением для регистрации данных времени.

⁽⁹⁾ Как и ранее, модель взаимодействия выполняется внутри нашего модуля model.

⁽¹⁰⁾ Если наша модель выдаёт отчёт об отказе для осуществления необходимого обновления, мы изменяем соответствующие данные отклика. Я включил это для тех читателей, которые ещё пока не знакомы с версией трёхместного оператора Python.

⁽¹¹⁾ Применяемые декораторы @app.listener являются предоставляемыми Sanic точками входа, которые дают вам какое- то место для добавления дополнительных действий в процессе последовательностей запуска и останова. Конкретно эта, before_server_start, вызывается перед тем как будет запущен наш сервер API. Она выглядит как подходящее место для инициализации нашего подключения к базе данных.

⁽¹²⁾ Мы применяем своего помощника Database для создания какого- то соединения с нашим экземпляром PostgreSQL. Той DB, к которой мы выполняем подключение, является restaurant.

⁽¹³⁾ Получаем некий пул подключений к нашей базе данных.

⁽¹⁴⁾ Применяем свою модель (к соответствующей таблице patron) чтобы создать необходимую таблицу если она отсутствует.

⁽¹⁵⁾ Применяем свою модель для создания некоей выделенной стороны ожидания для событий базы данных, причём ожидание мы выполняем в канале chan_patron. Соответствующей функцией обратного вызова для данных событий является model.db_event(), которую мы изучим в своём следующем листинге. Этот обратный вызов будет вызываться всякий раз, когда наша база данных обновляет соответствующий канал.

⁽¹⁶⁾ after_server_stop является той точкой входа, которая должна появиться в процессе останова. В этом месте мы отключаемся от своей базы данных.

⁽¹⁷⁾ Этот add_route() отправляет запросы POST для соответствующего URL /patron в нашу функцию сопрограммы new_patron().

⁽¹⁸⁾ Данный add_route() отправляет все запросы для соответствующего URL /patron/<id:int>> в представление на основе класса PatronAPI. Все названия методов в этом классе определяют именно то что они вызывают. Поэтому некий запрос GET вызовет соответствующий метод PatronAPI.get() и так далее.

⁽¹⁾ Чтобы быть способными получать уведомления при изменениях базы данных вам придётся добавить триггеры в свою базу данных. Я разработал эти полезные вспомогательные функции для создания необходимых функций триггера самих по себе (при помощи create_notify_trigger), а также для добавления такого триггера в некую определённую таблицу (посредством add_table_triggers). Необходимый для этого язык SQL является чем- то выходящим за рамки данной книги, тем не менее, это критически важно для понимания того как работает данный пример. Я включил соответствующий код с аннотациями для данных триггеров в соответствующее Дополнение.

⁽²⁾ Пакет стороннего разработчика boltons предоставляет некую пачку полезных инструментов, не последним из которых является собственно кэш LRU, более многостороннюю версию нежели наш декоратор @lru_cache из стандартного библиотечного модуля functools [Получите boltons воспользовавшись pip install boltons].

⁽³⁾ Данный блок текста содержит все необходимые SQL для наших стандартных операций CRUD. Отметим, что мы применяем стандартный синтаксис PostgreSQL для его параметров:$1, $2 и так далее. Здесь нет никакой новизны, ну и нечего обсуждать более.

⁽⁴⁾ Создаём необходимый кэш для данного экземпляра прикладного приложения.

⁽⁵⁾ Данная функция сопрограммы add_patron() является тем, что мы вызываем из своего модуля Sanic внутри соответствующей оконечной точки new_patron() для добавления новых покровителей. Внутри этой функции мы применяем для вставки новых данных метод fetchval() почему "fetchval", а не "fetchval"? Потому что "fetchval" возвращает значение первичного ключа вновь вставляемой записи! [Тем не менее, вам также потребуется соответствующая часть SQL RETURNING id!]

⁽⁶⁾ Обновляем некую имеющуюся запись. В случае успеха PostgreSQL возвратит UPDATE 1, следовательно мы используем это для проверки успешности данного обновления.

⁽⁷⁾ Удаление очень похоже на обновление.

⁽⁸⁾ Это операция "чтения". Именно она является той частью нашего интерфейса CRUD, которая требует позаботиться относительно состояния кэша. Задумайтесь об этом на мгновение: мы не обновляем значение кэша при осуществлении вставки, обновления или удаления. Это происходит потому, что мы полагаемся на существующие асинхронные уведомления от своей базы данных (через установленные триггеры) относительно обновлений значения кэширования в том случае когда изменяются любые данные.

⁽⁹⁾ Безусловно, мы всё ещё желаем применять кэширование после самого первого GET.

⁽¹⁰⁾ Наша функция db_event() является необходимым обратным вызовом, который asyncpg будет осуществлять когда имеются события в нашем канале уведомлений DB, chan_patron. Этот особый параметр списка требуемый со стороны asyncpg.conn является тем подключением, в которое были отправлены необходимые события, pid является значением идентификатора самого процесса того экземпляра PostgreSQL, который отправил данное событие, channel представляет собственно название нашего канала (и в нашем случае это chan_patron), а payload являются подлежащими отправке в этом канале данными.

⁽¹¹⁾ Выполняем разбор полученных данных JSON в некий dict.

⁽¹²⁾ Само заполнение кэша является достаточно прямолинейным, однако заметим, что события "обновления" содержат как новые, так и старые данные, поэтому нам необходимо гарантировать, что мы выполняем кэширование только новых данных.

⁽¹³⁾ Это небольшая функция утилиты, которую я создал для простого повторного создания таблицы при её отсутствии. Это действительно полезно ксли вам требуется часто выполнять данную операцию - например в случае написания примеров кода для данной книги! Именно в этом месте также создаются соответствующие коды триггеров уведомления нашей базы данных и добавляются в нашу таблицу patron. Относительно листинга этих функция с пояснениями обратитесь к нашему Дополнению.

Безусловно, наилучший рассказ относительно asyncio, с которым мне удалось познакомиться на YouTube, это Getting To Grips With Asyncio Роберта Смолшира, представленного на NDC в январе 2017 в Лондоне. Это выступление может быть достаточно сложным для начинающих, но на самом деле представляет ясное описание того как спроектирован asyncio.

Слайды Николая Новика, представленные на PyCon UA 2016: Building Apps With Asyncio. Эта информация очень плотная, но в этих слайдах содержится гигантский практический опыт.

Нескончаемые сеансы в Python REPL, в которых выполняются пробы м "наблюдается что происходит"!