Архитектор (JiaGouX) Мы все архитекторы!

Очередь задержки, как следует из названия, представляет собой очередь сообщений с функцией задержки. Итак, в каком случае мне нужна такая очередь?

задний план

Сначала рассмотрим следующие бизнес-сценарии:

Когда заказ не оплачен, как вовремя закрыть заказ и вернуть товар?
Как регулярно проверять, был ли успешно возвращен заказ в статусе возврата?
При создании нового магазина товар не загружался в течение N дней, откуда система узнает информацию и отправляет активационное смс? и т.д

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

Так есть ли готовое решение в мире открытого исходного кода? Ответ положительный. Beanstalkd (http://kr.github.io/beanstalkd/), который в основном соответствует вышеуказанным требованиям. Однако это не особенно удобно при удалении сообщений и требует больших затрат. Более того, он разрабатывался на основе языка C. В то время основным направлением работы нашей команды были PHP и Java, и не было возможности заниматься вторичной разработкой. Поэтому мы позаимствовали его дизайнерские идеи и повторно реализовали очередь с задержкой на Java.

Цели дизайна

Надежность передачи сообщения: после того, как сообщение попадает в очередь задержки, оно гарантированно будет использовано хотя бы один раз.
Расширенная поддержка клиентов: из-за бизнес-требований поддерживаются как минимум PHP и Python.
Высокая доступность: как минимум поддержка развертывания с несколькими экземплярами. После зависания экземпляра остаются резервные экземпляры для продолжения предоставления услуг.
В режиме реального времени: допустить определенную ошибку времени.
Поддержка удаления сообщений: бизнес-пользователи могут удалять указанные сообщения в любое время.

весь кадр

Вся очередь задержки состоит из 4 частей:

Пул заданий используется для хранения метаинформации обо всех заданиях.
Сегмент задержки — это набор упорядоченных очередей со временем в качестве измерения, используемый для хранения всех заданий, которые необходимо отложить/зарезервировать (здесь хранится только идентификатор задания).
Таймер отвечает за сканирование каждой корзины в режиме реального времени и помещение заданий, время задержки которых больше или равно текущему времени, в соответствующую очередь готовности.
Очередь готовности хранит задание в состоянии готовности (здесь хранится только идентификатор задания) для использования программой-потребителем.

Как показано на рисунке ниже:

有赞延迟队列设计

Очки дизайна

Базовые концепты

Задание: задача, которую необходимо обрабатывать асинхронно и которая является основной единицей в очереди задержки. Связан с определенной темой.
Тема: Набор (очередь) заданий одного типа. Чтобы потребители подписались.

структура сообщения

Каждое задание должно содержать следующие свойства:

Тема: Тип работы. Его можно понимать как конкретное название компании.
Id: уникальный идентификатор задания. Используется для извлечения и удаления указанной информации о задании.
Задержка: время, на которое задание должно быть отложено. Единица: секунды. (Сервер преобразует его в абсолютное время)
TTR (время выполнения): тайм-аут выполнения задания. Единица: секунды.
Тело. Содержимое задания, которое хранится в формате json, чтобы потребители могли выполнять конкретную бизнес-обработку.

Конкретная структура показана на следующем рисунке:

有赞延迟队列设计

TTR предназначен для обеспечения надежности передачи сообщений.

переход состояния сообщения

Каждое задание будет находиться только в определенном состоянии:

ready: исполняемое состояние, ожидание потребления.
задержка: неисполняемое состояние, ожидание тактов.
зарезервировано: было прочитано потребителем, но еще не получило ответа от потребителя (удалить, закончить).
удалено: было использовано или удалено.

Ниже приведена схема перехода четырех состояний:

有赞延迟队列设计

хранилище сообщений

Прежде чем выбрать носитель информации, сначала определите конкретную структуру данных:

Вакансии Опрос Работа метаинформация хранится, только K / V для формирования структуры. ключ для идентификатора задания, значение для структуры задания.
Delay Bucket — это упорядоченная очередь.
Готовая очередь — это обычный список или очередь.

Тот, кто может одновременно удовлетворить вышеуказанные требования, — это не что иное, как redis.
Структура данных корзины — это zset Redis, который разделен на несколько корзин для повышения скорости сканирования и уменьшения задержки сообщений.

письмо-соглашение

Чтобы соответствовать поддержке многоязычного Клиента, мы выбираем метод связи Http и реализуем взаимодействие с Клиентом через текстовый протокол (json). В настоящее время поддерживаются следующие протоколы:

Добавить: { 'команда': 'добавить', 'тема': 'xxx', 'id': 'xxx', 'задержка': 30, 'TTR': 60, 'тело': 'xxx'}
Получить: {'command': 'pop' ',' topic ':' xxx '}
Готово: {'команда': 'финиш', 'идентификатор': 'xxx'}
Удалить: {'команда':'удалить', 'идентификатор':'xxx'}

body также является строкой json.

Структура ответа: {'успех':правда/ложь,'ошибка':'причина ошибки','идентификатор':'xxx','значение':'тело задания'}
Подчеркните: идентификатор задания определяется бизнес-пользователем, и глобальная уникальность должна быть гарантирована. Рекомендуется использовать комбинацию темы и уникального идентификатора компании.

Пример жизненного цикла Job

Когда пользователь размещает заказ на определенный продукт, система успешно создает заказ и одновременно помещает задание в очередь задержки. Структура задания: {'тема':'заказзакрыть', 'id':'заказатьcloseorderNoXXX», «задержка»: 1800, «TTR»: 60, «тело»: «XXXXXXX»}
После того, как очередь задержки получает задание, она сначала сохраняет информацию о задании в пуле заданий, затем вычисляет абсолютное время выполнения в соответствии с задержкой и помещает идентификатор задания в корзину в циклическом режиме.
Таймер постоянно опрашивает каждую корзину. Через 1800 секунд (30 минут) он проверяет, истекло ли время выполнения вышеуказанного задания, и получает идентификатор задания, чтобы получить метаинформацию из пула заданий. Если задание в это время находится в удаленном состоянии, пропустите и продолжите опрос; если задание находится в неудаленном состоянии, сначала подтвердите, является ли задержка в метаинформации больше или равной текущему времени, и если да , поставить id задания в соответствующее готовое по теме Очередь затем удаляется из корзины, если она не устраивает, время задержки пересчитывается, снова помещается в корзину, а предыдущая работа id удален из ведра.
Потребитель опрашивает очередь готовности соответствующей темы (здесь еще надо судить о рациональности задания), а после получения задания делает свою бизнес-логику. При этом сервер пересчитывает время выполнения задания, полученного потребителем, в соответствии с его установленным TTR, и кладет его в корзину.
После обработки бизнеса потребитель отвечает на завершение сервера, и сервер удаляет соответствующую метаную информацию в соответствии с идентификатором задания.

Существующая физическая топология

有赞延迟队列设计

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

Неадекватный дизайн

Таймер реализован бесконечным циклом независимых потоков, что приведет к некоторой трате ресурсов процессора, когда нет готовой задачи.

Когда потребитель резервирует задание, он применяет метод короткого опроса HTTP, и одновременно может выполняться только одно задание. Если есть много готовых заданий, потребление сетевого ввода-вывода будет увеличено.
Redis используется для хранения данных, а сохраняемость сообщений ограничена характеристиками Redis.
Масштаб полагается на третьей стороне (Nginx).

будущее архитектурное направление

Реализация таймера на основе метода ожидания/уведомления.

Предоставляет API для длинных соединений TCP и реализует метод резервирования сообщений для принудительной отправки или длительного опроса.
Он имеет собственную схему хранения (встроенная база данных, настраиваемая структура данных для записи файлов) для обеспечения сохранности сообщений.

Реализуйте свой собственный сервер имен.

Рассмотрите возможность предоставления прямой поддержки для повторяющихся задач.

Источник: есть похожая технологическая команда

Оригинал: http: //tech.youzan.com/queuning_delay/

Если есть какие-либо нарушения или неуместность, пожалуйста, свяжитесь с Ruofei (WeChat: 1321113940), чтобы немедленно удалить его, спасибо!

·END·

архитектор

Мы все архитекторы!

Источник: общедоступный аккаунт WeChat: архитектор

пройти через"BeePress｜WeChat Official Account Article Collection" Создание плагина для WordPress