zset 설계 지연 큐 프로그램 기반 REDIS 도구 세트 #Redis

코드 저장소

background

하지만 일부 친구들은 사용 중 이 지연된 대기열 구성 요소에서 특히 메시지 손실과 같은 버그를 발견했습니다.

따라서 이를 대체할 다른 지연 대기열 솔루션이 필요하며, 현재 연구 끝에 다음 두 가지 아이디어를 사용할 수 있습니다:

지연된 대기열 메시징을 지원하는 새로운 메시징 미들웨어(예: Rocketmq)에 액세스합니다.
지연된 대기열 구성 요소를 자체 개발합니다(예: ZSET 구현을 기반으로).

사용법

종속성 소개

<dependency>
 <groupId>.org.wangchangjiu</groupId>
 <artifactId>redis-util-spring-boot-starter</artifactId>
 <version>1.0.1-SNAPSHOT</version>
</dependency>

구성 열기
지연된 메시지 보내기
지연된 메시지 수신
1. 자동 확인, 메서드에 @RedisDelayMessageListener 어노테이션을 추가하면 주제는 전송된 주제에 해당하며 자동 확인은 수신할 메시지 엔티티인 하나의 매개 변수만 사용할 수 있다는 점에 유의할 필요가 있습니다.
2. 수동 ACK, 메소드에 @RedisDelayMessageListener 주석을 추가하고 주제는 전송 된 주제에 해당하며 수동 ACK는 두 개의 매개 변수 여야하며 첫 번째 매개 변수는 수신 할 메시지 엔티티이고 두 번째 매개 변수는 메소드의 실행이 완료되거나 필요할 때 Acknowledgment로 고정되어 있어야합니다. 메서드가 실행되거나 필요할 때 acknowledgment.acknowledge()를 통해 수동으로 ACK를 제출하고, ACK가 제출되지 않으면 일정 시간이 지난 후 메시지를 지연 큐로 재전송합니다.

디자인 솔루션 순서도

집중 프로세스 설명

서비스가 시작되면 보스 스레드 풀, 작업자 스레드 풀, 백 스레드 풀의 세 가지 스레드 그룹이 시작됩니다;

보스 스레드에는 500밀리초마다 주기적으로 작업을 게시하여 지연 메시지를 받습니다;

워커 스레드는 다른 주제에 따라 지연된 메시지를 받으러 가고, 지연된 메시지를 받으면 리플렉션이 대상 메서드를 호출하여 실행하고, 실행이 성공하면 메시지를 자동/수동으로 승인합니다;

백 스레드는 5초마다 주기적으로 커밋되지 않은 ACK가 있는지 확인하고, 있는 경우 지연 대기열로 다시 전달합니다;
서비스가 시작되면 보스 스레드가 지연된 메시지를 일괄적으로 가져오는 작업을 수행하고, 워커 스레드가 지연된 메시지를 일괄적으로 가져오는 작업을 실행합니다. 이렇게 하는 이유는 서비스가 다운되거나 재시작될 때 많은 수의 지연된 메시지가 소비되지 않을 것을 우려하여 한 번에 일괄적으로 소비하기 위함입니다.
지연된 메시지 보내기 프로세스:
- 해시 데이터 구조에 저장되는 메시지 메타데이터를 추가합니다(키: task_meta_data:{서비스}).
- ZSET 데이터 구조에 메시지 ID를 추가하고, 값은 메시지ID, 점수는 현재 타임스탬프 + 만료 시간입니다.

구현 세부 정보

전체 개요

코어 빈 소개

지연 대기열 자동 어셈블리 빈: 지연 대기열 자동 어셈블리 빈입니다.

지연된 메시지 발신자: 지연된 메시지 발신자입니다.

DelayQueue: 지연 대기열, 메시지 보내기 / 메시지 받기 / 제출 ACK 코어 처리 클래스, 이 클래스는 redis + lua 스크립트를 캡슐화하여 redis 지연된 메시지를 처리하기 위해 lua 스크립트를 호출합니다;

위의 프로세스는 다음과 같이 설명됩니다.

컨테이너가 시작될 때의 주요 작업은 다음과 같습니다:
- 리플렉션 호출을 위한 대상 메서드 정보를 캡슐화하기 위해 메서드에 @RedisDelayMessageListener 어노테이션이 있는 빈을 확인합니다. 다음과 같이 다음 메서드 정보는 RedisDelayMessageConsumerContainer 객체에 캡슐화됩니다. 구현 코드:
- 보스 스레드 풀, 작업자 스레드 풀, 백 스레드 풀의 세 가지 스레드 그룹이 시작됩니다.
  - 보스 스레드에는 500밀리초마다 주기적으로 작업을 게시하여 지연 메시지를 받습니다;
  - 워커 스레드는 다른 주제에 따라 지연된 메시지를 받으러 가고, 지연된 메시지를 받으면 리플렉션이 대상 메서드를 호출하여 실행하고, 실행이 성공하면 메시지를 자동/수동으로 승인합니다;
  - 백 스레드는 5초마다 주기적으로 커밋되지 않은 ACK가 있는지 확인하고, 있는 경우 지연 대기열로 다시 전달합니다;
메시지 보내기

메시지 ID를 생성하고, 루아 스크립트를 호출하여 메시지를 추가합니다.
메시지 받기