Введение

📚 Общее количество персонажей: 6 тыс.+

⏳ Время чтения: 10 минут.

📢 Ключевые слова: колесо времени, двусвязный список, запланированная задача, Golang.

Тайминг Колесо) — Джордж Варгезе и Тони Lauckсуществовать1996годовые документы【Hashed and Hierarchical Timing Wheels: data structures to efficiently implement a timer facility】выполнитьиз,Он широко используется в ядре Linux.,Это один из методов калибровки и основы таймера ядра Linux.

По сравнению с собственным таймером и тикером Go, алгоритм временного колеса является более эффективной моделью планирования задач, подходящей для большего количества сценариев.

Однако временная точность кругового планирования может быть не очень высокой, и она может не подходить для задач планирования с особенно высокими требованиями к точности. Поскольку точность алгоритма колеса времени зависит от минимальной детализации «указателя» периода времени. Например, если сетка колеса времени прыгает один раз в секунду, задачи с точностью планирования менее одной секунды не могут быть запланированы по времени. колесо.

Применение колеса времени на самом деле очень обширно. Колесо времени можно найти в таких компонентах, как Netty, Akka, Quartz, ZooKeeper, Kafka и так далее.

Какую проблему решили

Если в системе имеется большое количество запланированных задач и если каждая из большого количества запланированных задач использует собственный планировщик для управления жизненным циклом задачи, это приведет к пустой трате ресурсов ЦП и будет очень неэффективным.

Колесо времени — это модель планирования, которая эффективно использует ресурсы потоков для пакетного планирования. Привяжите все большие пакеты запланированных задач к одному планировщику и используйте этот планировщик для управления, запуска и выполнения всех задач. Способен эффективно управлять различными отложенными задачами, периодическими задачами, задачами уведомлений и т. д.

принцип

Тайминг Wheel) — это кольцеобразная структура данных, похожая на часы, которую можно разделить на множество сеток (слотов). Нижний слой использует массив калибровки, каждая сетка представляет временной интервал. (Интервал) хранит определенный список запланированных задач (TaskList — это циклический двусвязный список). Каждый элемент связанного списка является запланированным заданием. Task。

одноступенчатое колесо

На картинке выше: это колесо времени с тиком 12, интервалом равным 1 с, currentTime = 3 и связанным списком со слотом = 3. В связанном списке 4 элемента задач.

Глядя на временную рулетку TimimgWheel под другим углом, можно просто понять ее как циферблат часов. Указатель время от времени перемещается на одну позицию. Круг — это цикл. Задачи, которые необходимо выполнить, расположены на шкале. циферблат. Когда указатель достигает этого положения, выполняется соответствующая задача.

Что будет после добавления задачи 5s?

Если текущий указатель currentTime указывает на 3, если в это время вставлена ​​задача на 5 с, новая задача будет сохранена во временной сетке 8.

колесо иерархии

Все вышеперечисленное представляет собой простые одноуровневые временные колеса. Если временной интервал проверяет слот шкалы временного колеса, например, добавление задачи, которая должна быть выполнена через 15 секунд, одно колесо не может удовлетворить наши потребности. Мы должны рассмотреть возможность использования других решений!

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

Мы предполагаем, что в одном раунде имеется m слотов, образующих массив (нижний уровень циклической очереди представлен массивом). Временной интервал, представленный каждым слотом, равен t, тогда временной диапазон, который может быть представлен, равен t. m * t и получите слот. Приведенный выше список задач может быть представлен как slot[i], где i — нижний индекс массива слотов.

Если это многоэтапный раунд, мы можем понимать его как несколько кольцевых очередей. Если двухэтапный раунд — это круг 1 и круг 2, если для круга 2 также выделено 12 слотов, временной интервал t каждого слота равен 1 с.

В это время временной диапазон, соответствующий каждому слоту в круге, должен составлять m * t (временной диапазон, представленный кругом1), а временной диапазон, который может представлять все вторичное колесо времени, равен m * m * t.

Почему указанный временной диапазон составляет m * m *t?

Поскольку круг2 основан на временном диапазоне m * t, выраженном кругом1, а время, выраженное слотом вторичного раунда, круг2 основан на круге, это также m слотов, поэтому весь диапазон представления равен m * m * t

По аналогии, если количество слотов 60, то выглядит ли этот метод как коэффициент преобразования времени, аналогичный [секундам, минутам, часам] Присмотритесь, это на самом деле так?

Так как же играть в несколько раундов? Вот два способа получить это требование

1: Метод мультирулетки

Концепция многоуровневого колеса времени ближе к часам, которые разделены на часовую, минутную и секундную стрелки. Каждые 60 секунд минутная стрелка перемещается на один кадр, а каждые 60 минут часовая стрелка перемещается на один кадр.

Идея многоуровневого временного колеса состоит в том, чтобы разделить рулетку на два или более уровней. Когда колесо рулетки первого уровня движется один раз, колесо рулетки второго уровня перемещается на одно пространство вперед после перемещения колеса рулетки второго уровня. примерно один раз, колесо рулетки третьего уровня перемещается на одну клетку вперед и так далее~.

2: круг ассоциации задач

В каждом задании добавляется параметр «круг» (количество кругов). Если количество кругов не превышает единой шкалы колеса времени, то число кругов «круг» = 0. Если оно превышает, вычисляется значение круга.

Например, задачу необходимо выполнить через 16 секунд, тогда круг задачи = 1, а рассчитанный слот положения циферблата = 4, то есть задачу необходимо выполнить в позиции 4 после того, как циферблат переместится один раз. Если указатель набора перемещается к слоту, кружок в списке задач уменьшается на 1, и задача не выполняется до тех пор, пока кружок задачи не станет = 0 в списке задач слота = 4.

Для сравнения, я лично предпочитаю добавлять в задачу параметр круга.,Что нужно контролировать таким образом, так это значение круга на задании.,Эта статья также основана на этом методе калибровки.

выполнить

После понимания принципа колеса времени,Давайте посмотрим, как использовать язык Golang для настройки колеса времени.,подизвыполнитьоснован на одном колесе、Объединяйте задачиcircleвыполнитьколесо иерархия, расширенное колесо времени, способ проверки.

Разберём процесс работы алгоритма колеса времени:

• Учреждатькольцевая структура данных（базовый массиввыполнить）
• Определите масштаб (слот) структуры данных, временной интервал, соответствующий каждому масштабу, и время, которое может представлять один круг, равное слоту * интервалу.
• Создайте задачу и определите слот шкалы и количество кругов, исходя из времени задержки выполнения задачи.
• Повесьте список запланированных задач для выполнения на весах
• Выполнение задач в соответствующем связанном списке слотов

идеи дизайна

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

Он разделен на три части: структура данных、Внешний подход、внутреннийвыполнить

Структуры данных — краеугольный камень Колеса Времени,На этих структурах основан весь цикл временного колеса.,Внешние методы — это то, что нам нужно использовать при использовании,Внутренняя — это адаптация конкретной бизнес-логики.,Среди них выполнение и getSlotAndCircle являются ядром всего решения по выбору.,Дальнейшие действия также являются предметом нашего разговора о проверке кода.

Вот две основы, основанные на использовании Golangвыполнения:

двусвязный список：на основеGolangизстандартная библиотекаcontainer/listуже ввыполнитьиздвусвязный список для выполнения базового хранения запланированных задач

кольцевая структура данных：здесьизкольцевая структура данных На самом деле нам не нужно осуществлять в практическом смыслеиз Концы соединяются, образуя кольцо.,Здесь основано на массиве,Затем используйте операцию остатка, чтобы пройти метод один за другим.,Цель восстановления, которая логически соответствует кольцевому подходу.

Начните произносить код восстановления!

основные структуры данных

Поле структуры временного колеса XTimingWhee представлено следующим рисунком:

Основные поля следующие:

currentSlot: текущая позиция в рулетке

слоты: используются для инициализации элементов в массиве рулетки. -- связанный список（на основемножествовыполнитьлогическиизкольцевая структура）

TaskRecords: это структура sync.Map, которая в основном отображает и управляет ключами задач и элементами, добавленными в связанный список, чтобы облегчить последующее удаление задач и другие операции оценки.

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

stopCha: выключить колесо времени

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

addTaskCh: основная структура запланированной задачи Task, которая будет обсуждаться позже.

RemoteTaskCh: просто передайте уникальный тег задачи.

Предупреждение о высоком энергопотреблении🔥🔥: Давайте поговорим о том, почему stopCh использует пустую структуру, поскольку для этого сценария выключения требуется только канал уведомлений, который не требует отправки каких-либо данных и не вызывает дополнительных затрат памяти.

Структура задачи

Определение структуры задачи следующее, и для XTimeingWheel нет информации о полях.

type Task struct {
    //Клавиша задачи только
    key       string
    //Конкретная функция, которую нужно выполнить
    job       Job
    //Время выполнения задачи
    executeAt time.Duration
    //Время выполнения -1: Повторить выполнение
    times     int
    // позиция в рулетке
    slot      int
    //Количество кругов
    circle    int
}

• ключ: уникальный идентификатор задачи, который должен быть глобально уникальным.
• задание: определена конкретная фактическая функция выполнения
• ExecuteAt: время задержки выполнения задачи
• Время: количество выполнений. Значение по умолчанию — -1, что означает повторное выполнение. Если оно равно 1, это означает одно выполнение.
• слот: позиция в рулетке, через которую можно получить доступ к круговому массиву путем подписки
• круг: количество кругов также можно назвать раундами. Если добавленные задачи не превышают количество колес рулетки, то значение равно 0. Если оно превышает: slotNum равен 12 и добавляются задачи, выполненные за 15 секунд. , затем круг = 1 и так далее.

Примечание. Задача может быть выполнена только тогда, когда круг = 0, что означает, что раунд пройден. Например, задача круга = 1 должна пройти один слой.

Его можно выполнить только тогда, когда значение круга равно 0.

Job — это фактическая функция выполнения,Да,Это функция, входной параметр которой является ключевым,Можетсуществоватьмы на самом делевыполнитьфункцияизвремя пришлотолько Пометить простую проверку,Структура его определения относительно проста:

type Job func(key string)

инициализация колеса времени

Перед использованием колеса времени необходимо его инициализировать. Здесь инициализация предоставляет два параметра:

интервал: временной интервал между каждым слотом колеса рулетки

slotNum: количество слотов в колесе рулетки.

func NewXTimingWheel(interval time.Duration, slotNum int) (*XTimeWheel, error) {

Существует также метод инициализации колеса синхронизации по умолчанию, DefaultTimingWheel, для которого по умолчанию интервал = 1, slotNum = 12.

func DefaultTimingWheel() (*XTimeWheel, error) {
	tw, _ := NewXTimingWheel(time.Second, DefaultSlotNum)
	return tw, nil
}

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

func NewXTimingWheel(interval time.Duration, slotNum int) (*XTimeWheel, error) {
        //Оценка параметра
	if interval <= 0 {
		return nil, errors.New("minimum interval need one second")
	}
	if slotNum <= 0 {
		return nil, errors.New("minimum slotNum need greater than zero")
	}
	t := &XTimeWheel{
		interval:     interval,
		currentSlot:  0,
		slotNum:      slotNum,
		slots:        make([]*list.List, slotNum),
		stopCh:       make(chan struct{}),
		removeTaskCh: make(chan string),
		addTaskCh:    make(chan *Task),
		isRun:        false,
	}
	t.start()
	return t, nil
}

func (t *XTimeWheel) start() {
        //Оцениваем, что колесо времени работает
	if !t.isRun {
                //Инициализируем двусвязную структуру массива по slotNum списокlist
		for i := 0; i < t.slotNum; i++ {
			t.slots[i] = list.New()
		}
                //Устанавливаем интервал таймера
		t.ticker = time.NewTicker(t.interval)
		t.mux.Lock()
		t.isRun = true
                //Начинаем выполнение сопрограммы
		go t.run()
		t.mux.Unlock()
	}
}

Мониторинг и распределение задач

Мы знаем, что Channel следует принципу «первым пришел — первым вышел» (First in First Out). In First Out) правила могут гарантировать порядок сбора. Эта функция также доставляет и распределяет сообщения, такие как добавление и удаление задач, в сочетании с + select Режим мультиплексирования для мониторинга нескольких channel Операции чтения и записи как простой уровень планирования.

Код выглядит следующим образом: Использовать for в качестве метода резидентной сопрограммы.

func (t *XTimeWheel) run() {
	for {
	select {
                //Закрыть колесо времени, Выйти из этой функции
		case <-t.stopCh:
			return
                //Добавить задачу
		case task := <-t.addTaskCh:
			t.addTask(task)
                //удаляем задачу
		case key := <-t.removeTaskCh:
			t.removeTask(key)
                //Сигнал таймера
		case <-t.ticker.C:
			t.execute()
		}
	}
}

Добавить задачу

Из карты разума проекта и метода запуска выберите,Можно предположить, что в добавлении задач фактически участвуют две стороны.,Давайте пройдемся по всему процессу:

• Внешний метод AddTask в основном используется для проверки параметров, а затем Task записывается в addTaskCh.
• Обработка события сигнала Добавить задачу в канале.
• Выполнение внутренних специфических задач, добавлена ​​оптимизация,Включает расчет слотов и кругов и добавление задач в двусвязный список указанного слота.

func (t *XTimeWheel) AddTask(key string, job Job, executeAt time.Duration, times int) error {
	if key == "" {
		return errors.New("key is empty")
	}
	if executeAt < t.interval {
		return errors.New("key is empty")
	}
        //sync.Map определяет, добавлен ли ключ задачи
	_, ok := t.taskRecords.Load(key)
	if ok {
		return errors.New("key of job already exists")
	}
	task := &Task{
		key:       key,
		job:       job,
		times:     times,
		executeAt: executeAt,
	}
  //Запись в канал addTaskCh
	t.addTaskCh <- task
	return nil
}

Внутренняя информация об AddTask следующая:

func (t *XTimeWheel) addTask(task *Task) {
        //Рассчитываем значения прорези и круга
        slot, circle := t.calSlotAndCircle(task.executeAt)
        task.slot = slot
        task.circle = circle
        //Сводка списка, добавленного к указанному индексу слота
        ele := t.slots[slot].PushBack(task)
        //sync.Map сохраняет ключевые и конкретные задачи в связанном списке
        t.taskRecords.Store(task.key, ele)
}

Расчет прорезей и кругов является основой постановки задачи на все колесо времени. Следующий раздел [Рассчитать прорезь и круг] будет представлен подробно.

Рассчитать прорезь и круг

Колесо времени будет рассчитывать слот и круг перед добавлением каждой задачи. Чтобы определить положение и количество кругов задачи в раунде, эти два параметра очень важны при планировании задач и будут основаны на текущем положении задачи. двусвязный список currentSlot. Выполнение таких операций, как циклические циклы.

func (t *XTimeWheel) calSlotAndCircle(executeAt time.Duration) (slot, circle int) {
        //Время задержки Второй
        delay := int(executeAt.Seconds())
        //Время, представленное текущим колесом рулетки Второй
        circleTime := len(t.slots) * int(t.interval.Seconds())
        //Рассчитываем количество кругов
        circle = delay / circleTime
        //Рассчитываем шаг слота, соответствующий времени задержки
        steps := delay / int(t.interval.Seconds())
        //Рассчитываем слот позиции
        slot = (t.currentSlot + steps) % len(t.slots)
        return
}

выполнять задания

Благодаря предыдущей инициализации Рассчитать прорезь и круг, после добавления задачи наша задача добавляется в связанный список. После срабатывания таймера в слоте, где находится задача, наступает этап выполнения. Вот самая важная бизнес-логика этой статьи. 🔥--[ выполнять задания】。

Давайте кратко рассмотрим: до сих пор задача добавлялась в двусвязный список, соответствующий указанной позиции, затем на этапе выполнения задачи из списка вынимаются и выполняются. Да, это ядро ​​всего. колесо времени посмотрим, как оно работает!

Сначала взгляните на код. Есть комментарии.

func (t *XTimeWheel) execute() {
        //Получаем соответствующий индекс слота list
	taskList := t.slots[t.currentSlot]
        //Определяем, пуст ли список
	if taskList != nil {
                //Обходим список
		for ele := taskList.Front(); ele != nil; {
                    //Получить запланированное задание элемента связанного списка
                    taskEle, _ := ele.Value.(*Task)
                    //Судить по кругу задач (circle == 0 выполняется)
                    if taskEle.circle > 0 {
			taskEle.circle--
                         //Вернем следующий элемент связанного списка
			ele = ele.Next()
			continue
			}
                    //выполнять заданияфункция                    go taskEle.job(taskEle.key)
            
                    //Удалить сопоставление клавиш delete
                    t.taskRecords.Delete(taskEle.key)
                    //удаляем задачу Местосуществоватьсвязанный списоксредний элемент
                    taskList.Remove(ele)
            
                    //Выполняем фиксированное количество задач
                    if taskEle.times-1 > 0 {
			taskEle.times--
			t.addTask(taskEle)
                    }
                    //повторяем задачу
                    if taskEle.times == -1 {
                        t.addTask(taskEle)
                    }
                    ele = ele.Next()
		}
	}
        //Увеличиваем слот текущей позиции вперед
	t.incrCurrentSlot()
}

После прочтения комментариев к коду вам должно быть все понятно. Если вам что-то непонятно, ничего страшного, у Сяо Сюя есть краткое изложение. 📝📝📝

1. Всякий раз, когда принимается сигнал тикерного таймера, список задач слотов будет получен посредством индексации массива на основе currentSlot.
2. Если часть списка currentSlot пуста, пройдите по списку, чтобы получить элементы в списке — запланированную задачу, которая будет выполнена.
3. Подожди, выполни задания, проверю параметры в задании кружок для проверки

• circle > Задача 0 еще не достигла времени выполнения, поэтому круг нужно Вычтите 1 и дождитесь обработки последующих раундов.
• circle = Задача 0 может быть выполнена в текущем раунде, запустив сопрограмму, выполните Функция, соответствующая заданию

1. Удалить текущую задачу после выполнения задачи
2. Когда параметр times задачи больше 0, times уменьшается на 1 при каждом выполнении, а times = -1 за повторяющиеся задачи Выполнить задания
3. Обход задачи текущего текущего слота завершен и введена следующая позиция рулетки.
4. Подождите, пока не сработает следующий сигнал синхронизации тикера.

другой

Выше мы разобрались с общим процессом и анализом кода. Студенты, прочитавшие схему, знают, что есть две части: удаление задач и остановка колеса времени. Конкретный код не будет опубликован.

Я разместил проект вgithubНа：https://github.com/xiaoxucode/xtimingwheel

Добро пожаловать, чтобы поставить звезду ⭐⭐

Помашите рукой Amway в конце статьи:

Друзья, приглашаем вас подписаться на мой одноимённый публичный аккаунт📢📢: [код Сяо Сюй], жду вас! 🤣🤣

👨👩Друзья, надеюсь, эта статья будет вам полезна~🌐

Добро пожаловать, ставьте лайк 👍, собирайте 💙, подписывайтесь 💡 и поддержите нас три раза подряд~🎈

🎈Чем больше ты знаешь, тем больше ты не знаешь. Я Сяо Сюй, увидимся в следующий раз~🙇💻.