13,8 тыс подписчиков

Как создать простой планировщик событий в Go

9 апреля 20219 апр 2021

3 мин

Базовый рабочий механизм представляет собой следующее: запланированное событие добавляется в базу данных, из которой оно будет выполняться в определенное время. Другая задача будет запускаться регулярно, чтобы проверить, не истекло ли какое-либо событие в базе данных. Если да, то запустится событие polling .

Начнем с создания схемы базы данных в postgresql, которая будет использоваться для хранения событий:

CREATE TABLE IF NOT EXISTS "public" ."jobs" (
"id" SERIAL PRIMARY KEY ,
"name" varchar (50 ) NOT NULL ,
"payload" text ,
"runAt" TIMESTAMP NOT NULL
)

Теперь определим структуру данных для следующих элементов.

Event — запланированная задача.
Listeners — список слушателей событий.
ListenFunc — функция, которая будет вызываться при запуске события.

// Listeners прикрепляет слушателей событий type Listeners map[string ]ListenFunc

// Функция ListenFunc, которая прослушивает события type ListenFunc func (string )

// Структура события type Event struct {
ID uint Name string Payload string }

Также определим структуру Scheduler , которая будет планировать события и запускать слушателей:

// Структура данных планировщика type Scheduler struct { db *sql.DB
listeners Listeners
}

// NewScheduler создает новый планировщик func NewScheduler (db *sql.DB, listeners Listeners) Scheduler {
return Scheduler{
db: db,
listeners: listeners,
}
}

Здесь мы создаем новый планировщик, передавая ему экземпляр sql.DB и начальных слушателей.

Теперь нужно добавить реализацию функции планирования, которая будет помещать событие в таблицу jobs :

В функции AddListener мы просто присваиваем функцию listener к имени события.

Мы завершили первый этап: добавили событие в таблицу job . Теперь нужно извлечь устаревшие задачи из базы данных, выполнить их, а затем удалить.

Реализация функции ниже показывает, как можно выявить устаревшие события в таблице, а также сериализацию события в структуру Event :

Переходим ко второму этапу: вызываем зарегистрированных слушателей событий из базы данных:

Здесь мы проверяем наличие функции event . Если она прикреплена, то мы вызываем функцию event listener . Строки 6–9 удаляют задачу, поэтому при повторном поиске по базе данных слушатель не будет найден.

И наконец, переходим к финальному этапу: проверяем, истекло ли какое-либо событие в заданный интервал времени. Для запуска задач в определенный период используем функцию ticker библиотеки time , которая предоставит канал, получающий новый тик в заданном интервале.

Здесь мы проверяем, закрыт ли контекст, или же канал ticker получает тики. После получения тиков просматриваем соответствующие события, а затем вызываем слушателей для всех событий.

Теперь мы будем использовать все функции, определенные ранее в файле main.go :

В строках 13–16 мы прикрепляем слушателей к имени событий SendEmail и PayBills , чтобы эти функции вызывались при появлении новой задачи.

В строках 22 и 32–37 мы прикрепляем канал прерывания (interrupt ) с помощью os.Interrupt . Когда прерывание в программе выполняется, мы отменяем контекст в строке 19.

В строках 26–30 мы определяем планировщик событий, запускаем функцию polling и планируем выполнение события SendEmail через минуту, а PayBills — через две минуты.

Вывод данной программы выглядит следующим образом:

Мы видим, что событие SendEmail было выполнено через минуту, а PayBills — в следующую минуту.

Таким образом, мы создали базовую систему планирования событий, которая выполняет задачи в определенный временной интервал. Здесь можно найти полный пример кода.