DelayQueue — очередь отложенных сообщений

Введение

Архитектура: SpringCloud-Stream + RocketMQ + RocksDB (можно использовать Nacos, Eureka в качестве регистрационного центра, как отдельный сервис отложенных сообщений).

Это разновидность очереди отложенных сообщений на основе Apache RocketMQ.

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

Связанные с Apache RocketMQ очереди сообщений: TOPIC_SYSTEM_QUEUE_REALTIME: очередь реальных сообщений (когда сообщение истекает, количество сообщений в этой очереди увеличивается на 1); TOPIC_SYSTEM_QUEUE_DELAY: очередь отложенных запросов (используется для приёма запросов на отложенные сообщения).

Поддержка функций версии 1.0.0:

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

Процесс запроса:

1. Приложение вызывает запрос.
2. Отправляется отложенное сообщение.
3. Приложение записывает сообщение в TOPIC_SYSTEM_QUEUE_DELAY.
4. Отложенный сервер прослушивает TOPIC_SYSTEM_QUEUE_DELAY, считывает сообщения и записывает их в RocksDB.
5. Сервер с задержкой циклически обходит RocksDB, проверяя, нужно ли отправлять отложенные сообщения.
6. Если сообщение необходимо отправить, оно помещается в TOPIC_SYSTEM_QUEUE_REALTIME.
7. Приложение отслеживает TOPIC_SYSTEM_QUEUE_REALTIME для выполнения бизнес-операций.

Как расширить?

Полностью наследует высокую доступность, высокую производительность, низкую задержку и надёжность Apache RocketMQ. Если производительности RocketMQ недостаточно, добавьте узлы брокера RocketMQ. Если потребление сообщений недостаточно, скопируйте эту услугу и настройте группу потребления на TOPIC_SYSTEM_QUEUE_DELAY. Также можно преобразовать её в службу Eureka или Nacos, чтобы обеспечить поддержку высокой доступности Restful.

Как вызвать

Предоставляются два режима: один — интерфейс Restful, другой — SpringCloudStream.

Режим интерфейса Restful:

Незадержанное сообщение:

GET 1.1 http://delay-server:port/message/push/normal

Параметры: id: идентификатор сообщения (уникальный); message: содержимое сообщения. Пример: GET 1.1 http://delay-server:port/message/push/normal?id=1&message=Hello World

Задержанное сообщение (отметка времени):

GET 1.1 http://delay-server:port/message/push/delay

Параметры: id: идентификатор сообщения (уникальный); message: содержимое сообщения; timestamp: время отправки. Пример: GET 1.1 http://delay-server:port/message/push/delay?id=1&message=Hello World&timestamp=метка времени отправки сообщения в будущем

Отложенное сообщение (строка даты):

GET 1.1 http://delay-server:port/message/push/delay/pattern

Параметры: id: идентификатор сообщения (уникальный); message: содержимое сообщения; date: строка даты; pattern: правило форматирования. Пример: GET 1.1 http://delay-server:port/message/push/delay/pattern?id=1&message=Hello World&date=2021-03-03 11:00:00&pattern=yyyy-MM-dd HH:mm:ss

Режим SpringCloudStream

Настройте канал производства сообщений Topic, Topic TOPIC_SYSTEM_QUEUE_DELAY

spring:
  cloud:
    stream:
      rocketmq:
        binder:
          # Используйте этот адрес в новой версии
          name-server: rocketmq-server:9876
      bindings:
        # Канал отправки
        system_topic_queue_realtime_output:
          # Пункт назначения: тема (Topic)
          destination: TOPIC_SYSTEM_QUEUE_REALTIME
        # Канал отправки
        system_topic_queue_delay_output:
          # Пункт назначения: тема (Topic)
          destination: TOPIC_SYSTEM_QUEUE_DELAY
        # Приёмный канал
        system_topic_queue_delay_input:
          # Пункт назначения: тема (Topic)
          destination: TOPIC_SYSTEM_QUEUE_DELAY
          # Тип MessageConvert, протокол JSON
          content-type: application/json
          # Группа потребления
          group: TOPIC_SYSTEM_QUEUE_DELAY_GROUP
  application:
    name: omuao-delay-queue
  jackson:
    date-format: yyyy-MM-dd HH:mm:ss
    time-zone: GMT+8
    default-property-inclusion: non_null
  mvc:
    date-format: yyyy-MM-dd HH:mm:ss
  servlet:
    multipart:
      max-file-size: 10MB
      max-request-size: 10MB