Конкурентність (runner)

Цей пакет можна використовувати, якщо ви запускаєте бота використовуючи тривале опитування та хочете, щоб повідомлення оброблялися паралельно.

Переконайтеся, що ви розумієте 2-й етап масштабування, перш ніж почнете використовувати runner.

Навіщо потрібен runner

Якщо ви розміщуєте бота використовуючи тривале опитування та хочете його масштабувати, не можна обійтися без паралельної обробки оновлень, оскільки послідовна обробка оновлень відбувається надто повільно. Унаслідок цього боти стикаються з низкою проблем.

• Чи існують стани гонитви?
• Чи можемо ми все ще чекати (await) виконання стеку проміжних обробників? Ми мусимо знати це для обробки помилок!
• Що робити, якщо проміжний обробник з якихось причин ніколи не виконається? Чи заблокує це бота?
• Чи можемо ми обробити деякі вибрані оновлення послідовно?
• Чи можемо ми обмежити навантаження на сервер?
• Чи можна обробляти оновлення на декількох ядрах?

Як бачите, нам потрібне рішення, яке може вирішити всі перераховані вище проблеми, щоб досягти належного тривалого опитування для бота. Це проблема, яка дуже відрізняється від написання проміжного обробника або надсилання повідомлень у Telegram. Отже, вона не вирішується за допомогою базового пакету grammY. Замість цього ви можете використовувати runner. Він також має власну довідку API.

Використання

Ось простий приклад.

TypeScript

import { Bot } from "grammy";
import { run } from "@grammyjs/runner";

// Створюємо бота.
const bot = new Bot("");

// Додаємо звичайний проміжний обробник тощо
bot.on("message", (ctx) => ctx.reply("Отримав ваше повідомлення."));

// Запускаємо бота з конкурентним виконанням!
run(bot);

JavaScript

const { Bot } = require("grammy");
const { run } = require("@grammyjs/runner");

// Створюємо бота.
const bot = new Bot("");

// Додаємо звичайний проміжний обробник тощо
bot.on("message", (ctx) => ctx.reply("Отримав ваше повідомлення."));

// Запускаємо бота з конкурентним виконанням!
run(bot);

Deno

import { Bot } from "https://deno.land/x/grammy@v1.27.0/mod.ts";
import { run } from "https://deno.land/x/grammy_runner@v2.0.3/mod.ts";

// Створюємо бота.
const bot = new Bot("");

// Додаємо звичайний проміжний обробник тощо
bot.on("message", (ctx) => ctx.reply("Отримав ваше повідомлення."));

// Запускаємо бота з конкурентним виконанням!
run(bot);

Послідовна обробка там, де це необхідно

Швидше за все, ви хочете бути впевненими, що повідомлення з одного чату будуть оброблятися послідовно. Це корисно при встановленні проміжного обробника сесії, але це також гарантує, що ваш бот не переплутає порядок повідомлень в одному чаті.

Runner експортує проміжний обробник sequentialize, який піклується про це. Ви можете переглянути цей розділ, щоб дізнатися, як ним користуватися.

Зараз ми розглянемо більш просунуте використання плагіна.

Функція обмеження, що постачається з плагіном, може бути використана не лише для визначення ідентифікатора чату або ідентифікатора користувача. Замість цього ви можете повернути список рядків ідентифікаторів обмежень, які окремо для кожного оновлення визначають, на які ще обчислення воно має зачекати, перш ніж почне обробку.

Наприклад, ви можете повернути ідентифікатор чату й ідентифікатор користувача автора повідомлення.

bot.use(
  sequentialize((ctx) => {
    const chat = ctx.chat?.id.toString();
    const user = ctx.from?.id.toString();
    return [chat, user].filter((con) => con !== undefined);
  }),
);

Це гарантує, що повідомлення в одному чаті будуть впорядковані правильно. Крім того, якщо Аліса надсилає повідомлення в групі, а потім надсилає повідомлення вашому боту в приватному чаті, то ці два повідомлення будуть впорядковані правильно.

У певному сенсі, ви можете задати граф залежностей між оновленнями. Runner вирішить всі необхідні обмеження на льоту і заблокує ці оновлення до тих пір, поки це буде необхідно для забезпечення правильної послідовності повідомлень.

Реалізація цього дуже ефективна. Вона потребує константної памʼяті, якщо ви не вкажете нескінченну конкурентність, та сталого часу на обробку кожного оновлення.

Коректне завершення роботи

Для того, щоб бот коректно завершив свою роботу, ви повинні повідомити йому, щоб він зупинився, коли процес наближається до завершення.

Зауважте, що ви можете дочекатися завершення роботи runner’а, чекаючи (await) завдання (task) у RunnerHandle, повернутого з run.

const handle = run(bot);

// Він буде викликаний, коли бот зупиниться.
handle.task().then(() => {
  console.log("Бот завершив обробку!");
});

// Пізніше зупиніть бота за допомогою керуючого елементу runner'а.
await handle.stop();

Просунуті параметри

Runner складається з трьох частин: джерела оновлень (source), поглинача оновлень (sink) та runner’а. Джерело надає оновлення, поглинач обробляє оновлення, а runner налаштовує та зʼєднує ці дві частини.

Детальний опис того, як працює runner, можна знайти тут.

Кожну з цих трьох частин можна налаштувати за допомогою різних опцій. Завдяки цьому можна зменшити мережевий трафік, вказати дозволені оновлення тощо.

Кожна частина runner’а отримує свої налаштування за допомогою спеціального обʼєкта options.

run(bot, {
  source: {}, // параметри джерела оновлень
  runner: {}, // параметри runner'а
  sink: {}, // параметри поглинача оновлень
});

Вам слід переглянути RunOptions у довіднику API, щоб дізнатися, які параметри доступні.

Наприклад, там ви дізнаєтеся, що allowed_updates можна увімкнути за допомогою наступного фрагмента коду.

run(bot, { runner: { fetch: { allowed_updates: [] } } });

Багатопоточність

Якщо ваш бот не обробляє щонайменше 50 мільйонів оновлень на день (>500 на секунду), у багатопоточності немає сенсу. Пропустіть цей розділ, якщо ваш бот обробляє менше трафіку.

JavaScript є однопоточним. Це дивовижно, тому що паралелізм складний, а це означає, що якщо є лише один потік, знімається багато головного болю.

Однак, якщо ваш бот має надзвичайно високе навантаження (ми говоримо про 1000 оновлень в секунду і вище), то виконання всього на одному ядрі може виявитися недостатньо. По суті, єдине ядро намагатиметься впоратися з обробкою JSON для всіх повідомлень, які має обробити ваш бот.

Worker’и бота для обробки оновлень

Існує простий вихід: worker’и бота! Runner дозволяє створювати декілька worker’ів, які можуть обробляти ваші оновлення паралельно на різних ядрах, використовуючи різні цикли подій (event loop) та окрему памʼять.

У Node.js runner використовує Worker Threads. У Deno runner використовує Web Workers.

Концептуально, runner надає вам клас під назвою BotWorker, який може обробляти оновлення. Він еквівалентний звичайному класу Bot: фактично, він навіть розширює (extends) клас Bot. Основна відмінність між BotWorker і Bot полягає в тому, що BotWorker не може отримувати оновлення. Замість цього він повинен отримувати їх від звичайного Bot, який керує своїми worker’ами.

1. отримує оновлення                             Bot
                                              __// \\__
                                           __/  /   \  \__
2. надсилає оновлення worker'ам         __/    /     \    \__
                                     __/      /       \      \__
                                    /        /         \        \
3. обробляють оновлення     BotWorker   BotWorker   BotWorker   BotWorker

Runner надає вам проміжний обробник, який може надсилати оновлення worker’ам. Worker’и отримують ці оновлення та обробляють їх. Отже, центральному боту потрібно лише отримувати та розповсюджувати оновлення серед ботів, якими він керує. Фактична обробка оновлень: фільтрація повідомлень, надсилання відповідей тощо, виконується worker’ами бота.

Давайте подивимося, як це можна використовувати.

Використання worker’ів бота

Приклади цього можна знайти у репозиторії runner’а.

Ми почнемо зі створення центрального екземпляра бота, який отримуватиме оновлення та передаватиме їх worker’ам. Почнемо зі створення файлу з назвою bot.ts з наступним вмістом.

TypeScript

// bot.ts
import { Bot } from "grammy";
import { distribute, run } from "@grammyjs/runner";

// Створюємо бота.
const bot = new Bot(""); // <-- Помістіть токен свого бота між ""

// За бажанням, можемо впорядкувати оновлення.
// bot.use(sequentialize(...))

// Передаємо оновлення worker'ам бота.
bot.use(distribute(__dirname + "/worker"));

// Запускаємо бота з паралельним виконанням.
run(bot);

JavaScript

// bot.js
const { Bot } = require("grammy");
const { distribute, run } = require("@grammyjs/runner");

// Створюємо бота.
const bot = new Bot(""); // <-- Помістіть токен свого бота між ""

// За бажанням, можемо впорядкувати оновлення.
// bot.use(sequentialize(...))

// Передаємо оновлення worker'ам бота.
bot.use(distribute(__dirname + "/worker"));

// Запускаємо бота з паралельним виконанням.
run(bot);

Deno

// bot.ts
import { Bot } from "https://deno.land/x/grammy@v1.27.0/mod.ts";
import { distribute, run } from "https://deno.land/x/grammy_runner@v2.0.3/mod.ts";

// Створюємо бота.
const bot = new Bot(""); // <-- Помістіть токен свого бота між ""

// За бажанням, можемо впорядкувати оновлення.
// bot.use(sequentialize(...))

// Передаємо оновлення worker'ам бота.
bot.use(distribute(new URL("./worker.ts", import.meta.url)));

// Запускаємо бота з паралельним виконанням.
run(bot);

Поруч з bot.ts створюємо другий файл під назвою worker.ts, як зазначено у 12-му рядку в коді вище. Він міститиме власне логіку роботи бота.

TypeScript

// worker.ts
import { BotWorker } from "@grammyjs/runner";

// Створюємо нового worker'а бота
const bot = new BotWorker(""); // <-- передайте токен бота знову

// Додаємо логіку обробки повідомлень.
bot.on("message", (ctx) => ctx.reply("Еге-гей!"));

JavaScript

// worker.js
const { BotWorker } = require("@grammyjs/runner");

// Створюємо нового worker'а бота
const bot = new BotWorker(""); // <-- передайте токен бота знову

// Додаємо логіку обробки повідомлень.
bot.on("message", (ctx) => ctx.reply("Еге-гей!"));

Deno

// worker.ts
import { BotWorker } from "https://deno.land/x/grammy_runner@v2.0.3/mod.ts";

// Створюємо нового worker'а бота
const bot = new BotWorker(""); // <-- передайте токен бота знову

// Додаємо логіку обробки повідомлень.
bot.on("message", (ctx) => ctx.reply("Еге-гей!"));

Зверніть увагу, що кожен worker може надсилати повідомлення назад у Telegram. Ось чому ви також повинні передати токен бота кожному worker’у.

Вам не потрібно запускати worker’ів бота або експортувати щось з файлу. Достатньо створити екземпляр BotWorker. Він буде автоматично обробляти оновлення.

Важливо розуміти, що лише необроблені оновлення надсилаються worker’вм бота. Іншими словами, обʼєкти контексту створюються двічі для кожного оновлення: перший раз в bot.ts, щоб його можна було передати worker’у бота, а другий раз в worker.ts, щоб його можна було обробити. Крім того, властивості, встановлені для обʼєкта контексту в bot.ts, не надсилаються worker’ам бота. Це означає, що всі плагіни повинні бути встановлені у worker’ах бота.

Передача лише певних оновленнь

Для оптимізації продуктивності ви можете відкидати оновлення, які не хочете обробляти. Тож вашому боту не доведеться надсилати оновлення worker’у, щоб той його проігнорував.

Node.js

// Бот оброблятиме лише звичайні й редаговані повідомлення та запити зворотного виклику,
// тому ми можемо ігнорувати всі інші оновлення й не передавати їх.
bot.on(
  ["message", "edited_message", "callback_query"],
  distribute(__dirname + "/worker"),
);

Deno

// Бот оброблятиме лише звичайні й редаговані повідомлення та запити зворотного виклику,
// тому ми можемо ігнорувати всі інші оновлення й не передавати їх.
bot.on(
  ["message", "edited_message", "callback_query"],
  distribute(new URL("./worker.ts", import.meta.url)),
);

Початково distribute створює 4 worker’и бота. Ви можете легко змінити цю кількість.

// Передаємо оновлення 8-и worker'ам бота.
bot.use(distribute(workerFile, { count: 8 }));

Зауважте, що ваша програма ніколи не повинна створювати більше потоків, ніж є фізичних ядер вашого процесора. Це не покращить продуктивність, а навпаки погіршить її.

Як це працює за кулісами

Хоча використання runner виглядає дуже просто, під капотом відбувається багато чого.

Кожен runner складається з трьох різних частин.

1. Джерело підтягує оновлення з Telegram.
2. Поглинач надає оновлення екземпляру бота.
3. Компонент runner зʼєднує джерело і поглинач, а також дозволяє запускати і зупиняти бота.

api.telegram.org <—> джерело <—> runner <—> поглинач <—> бот

Джерело

Runner постачається з одним джерелом за замовчуванням, яке може працювати з будь-яким UpdateSupplier (довідка API). Такий постачальник оновлень легко створити з екземпляра бота. Якщо ви хочете створити його самостійно, обовʼязково ознайомтеся з createUpdateFetcher (довідка API).

Джерело — це асинхронний ітератор пакетів оновлень, але він може бути активним або неактивним і ви можете закрити (close) його, щоб відʼєднатися від серверів Telegram.

Поглинач

Runner постачається з трьома можливими реалізаціями поглинача: послідовною, у якої поведінка така сама, як у bot.start(), пакетною, яка переважно корисна для зворотної сумісності з іншими фреймворками, та повністю паралельною, яка використовується у методі run. Всі вони працюють з обʼєктами UpdateConsumer (довідка API), які легко створити з екземпляра бота. Якщо ви хочете створити його самостійно, обовʼязково перегляньте handleUpdate на екземплярі класу Bot (довідка API).

Поглинач містить чергу (довідка API) окремих оновлень, які наразі обробляються. Додавання нових оновлень до черги негайно змусить поглинача оновлень обробити їх і повернути Promise, який вирішиться, як тільки в черзі знову зʼявиться вільне місце. Розвʼязане ціле число визначає вільний простір. Отже, runner дотримується обмеження на паралельну обробку через екземпляр черги, що лежить в основі.

Черга також викидає оновлення, обробка яких займає занадто багато часу; ви можете вказати timeoutHandler при створенні відповідного поглинача. Звісно, при створенні поглинача також слід передбачити обробник помилок.

Якщо ви використовуєте run(bot), буде використано обробник помилок з bot.catch.

Runner

Runner — це звичайний цикл, який витягує оновлення з джерела та надає їх поглиначу. Як тільки у поглинача знову зʼявиться вільне місце, runner отримає наступну порцію оновлень з джерела.

Коли ви створюєте runner за допомогою createRunner (довідка API), ви отримуєте обʼєкт, який можна використовувати для керування runner’ом. Наприклад, ви можете запускати і зупиняти його або отримувати Promise, який виконується у разі зупинки runner’а. Цей обʼєкт також повертається методом run. Ознайомтеся з довідкою API обʼєкта RunnerHandle.

Функція run

Функція run робить кілька речей, які допоможуть вам легко використовувати наведену вище структуру.

1. Вона створює постачальника оновлень з вашого бота.
2. Створює джерело від постачальника оновлень.
3. Створює споживача оновлень з вашого бота.
4. Створює поглинача зі споживача оновлень.
5. Створює runner з джерела та поглинача.
6. Запускає runner.

До того ж повертається обʼєкт створеного runner’а, який дозволяє вам керувати runner’ом.

Загальні відомості про плагін

• Назва: runner
• Джерело
• Довідка