Redis의 메시지 큐 및 지연된 메시지 큐 구현 방법에 대한 간략한 설명

Redis는 메시지 대기열과 지연된 메시지 대기열을 어떻게 구현합니까? 다음 글에서는 Redis에서 메시지 큐와 지연된 메시지 큐를 구현하는 방법을 소개하겠습니다. 도움이 되셨으면 좋겠습니다!

Redis의 경우 더 많은 사람들이 이를 캐시로 사용한다고 생각할 수 있습니다. 실제로 Redis는 목록 데이터 구조를 사용하여 대기열을 구현할 수도 있습니다. [관련 권장 사항: Redis 동영상 튜토리얼]

list의 여러 명령

lpush(왼쪽 푸시)

은 대기열의 왼쪽에 저장됩니다.

rpush(오른쪽 푸시)

는 오른쪽에 저장됩니다. side of the queue

lpop(왼쪽 팝)

큐 왼쪽에서 꺼내기

rpop(오른쪽 팝)

큐 오른쪽에서 꺼내기

위의 네 가지 명령으로 목록을 만들 수 있습니다. 큐 또는 스택을 구현하는 데 도움이 되며 큐의 특성은 선입선출이고 스택의 특성은 선입선출입니다.

그래서 큐 구현은 lpush + rpop 또는 rpush + lpop을 사용할 수 있습니다.

스택 구현은 lpush + lpop 또는 rpush + rpop입니다.

대기열을 보여주기 위해 명령을 사용하세요

생산자가 메시지를 게시합니다

먼저 rpush를 사용하여 inform-queue라는 대기열에 다섯 가지 요소, 즉 1 2 3 4 5를 추가합니다. 즉, 생산자로서 뉴스를 게시합니다

소비자는 뉴스를 소비합니다

생산자는 rpush를 사용하므로 소비자는 lpop을 사용해야 합니다. 아래 그림을 보면 계속해서 통지 대기열이 표시됩니다. 1부터 5까지 순서대로 진행되고, 결국 큐에 메시지가 없고, 팝업이 항상 비어있습니다

빈 폴링 문제

lpop을 사용하세요.

위는 수동 실행 명령인데, 코드를 작성하면 프로그램이 계속 진행됩니다. 팝 데이터(데이터 가져오기)는 빈 폴링(쓸모 없는 읽기)을 유발하여 클라이언트의 CPU 소비를 증가시킬 뿐만 아니라 Redis의 QPS도 증가시키며 여전히 쓸모 없는 작업일 수 있습니다. 결과적으로 다른 클라이언트의 Redis 액세스 응답 속도가 느려집니다.

솔루션 A(sleep)빈 폴링은 클라이언트와 Redis의 리소스 소비를 증가시키므로 클라이언트가 빈 데이터 Sleep을 수신할 때 1초를 수행한 다음 1초 후에 데이터를 가져오도록 하여 소비를 줄일 수 있습니다.

Thread.sleep(1000)이 솔루션에도 단점이 있습니다. 즉, 컨슈머가 1명만 있으면 지연이 1초, 즉 빈 폴링이 발생하게 되는데, 이때는 잠자기 상태가 됩니다. 뉴스가 있는데 소비하기 전에 깨어나려면 여전히 1초까지 기다려야 합니다.

소비자가 여러 명인 경우 각 소비자의 수면 시간이 나누어지기 때문에 약간의 대기 시간이 줄어들지만 이를 달성하는 더 좋은 방법이 있습니까? 거의 0 레이턴시?

솔루션 B(읽기 차단) 실제로 Redis에는 대기열 데이터 가져오기에 대한 두 가지 명령이 있습니다. 즉, 읽기 차단,

blpop(왼쪽 팝 차단)

brpop(오른쪽 팝 차단)

읽기 차단입니다. 대기열에 데이터가 없으면 휴면 상태로 전환됩니다. 메시지가 오면 즉시 응답하고 데이터를 읽습니다. 따라서 lpop/rpop을 대체하기 위해 blpop/brpop을 사용하면 메시지 지연 문제를 해결할 수 있습니다. 큐에 6, 7, 8

3가지 속성을 추가하려면 blpop을 사용하여 큐를 읽으세요. 마지막 매개변수는 읽기를 차단하는 대기 시간입니다. 이 시간 이후에 메시지가 없으면 nil이 반환됩니다. 이때 blpop 작업을 계속 반복하면 됩니다.

읽기 차단을 위한 유휴 연결 자동 연결 끊김 문제

클라이언트가 읽기 차단을 사용할 때 차단 시간이 너무 길면 서비스가 중단됩니다. 일반적으로 이를 유휴 연결로 처리하므로 리소스를 차지하는 불필요한 연결을 줄이기 위해 적극적으로 연결을 끊습니다. 이때 클라이언트는 예외를 발생시킵니다. 따라서 클라이언트가 차단 읽기를 사용할 경우 예외를 캡처해야 합니다. 다시 시도하는 등 적절하게 처리하십시오.

java 클라이언트는 메시지 큐를 구현합니다

명령줄 클라이언트 redis-cli가 Java 언어로 변경되고 하나의 스레드 또는 여러 스레드가 rpush 게시를 수행한다는 점을 제외하면 아이디어는 위와 동일합니다.

또 다른 하나 이상의 스레드가 blpop 소비를 수행합니다. 완성된 코드는 https://github.com/qiaomengnan16/redis-demo/tree/main/redis-queue

Publisher

구독 에 있습니다.

지연 대기열 구현 아이디어

지연 대기열은 메시지가 전송된 후가 아니라 일정 시간이 지난 후 소비자가 메시지를 소비한다는 의미로 소비자가 즉시 읽을 수 있습니다.

zset은 이를 수행하는 데 도움이 될 수 있습니다. 먼저 zset은 점수별로 정렬할 수 있고 점수는 타임스탬프를 저장할 수 있으므로 메시지를 게시할 때마다 현재 타임스탬프와 지연된 타임스탬프를 사용합니다.

그런 다음 소비자는 메시지를 받고 zset의 데이터를 가로채서 현재 시간을 충족한 메시지를 가져옵니다(즉, 현재 타임스탬프보다 작거나 같은 점수의 데이터를 가져옵니다. 현재 타임스탬프보다 작거나 같은 점수는 의미합니다). 메시지가 해당 시간에 도달했다는 의미입니다. 이 값이 크면 소비되기까지 잠시 기다려야 한다는 의미입니다.

주요 명령 zadd(게시자), zrangebyscore(구독자), zrem(구독자가 데이터 소비 후 삭제)

명령 구현

zadd를 사용하여 4개의 데이터, 즉 1과 2를 추가했습니다. 3초 후에 소비되고(유사하게 말하면 이것은 실제로 점수일 뿐입니다), 10초 후에 소비될 수 있는 kafka,

3초에 도달하면 더 큰 값을 zset에서 취합니다. 1보다 작거나 같음 초의 합은 3초의 데이터보다 작거나 같습니다. 왜냐하면 이 범위의 데이터가 우리가 소비할 수 있는 것과 정확히 같기 때문입니다. 보시다시피 조건을 충족하는 데이터 3개를 꺼냈습니다. ,

한 번에 하나의 데이터만 소비할 수 있는 경우 제한 제한 조건을 추가할 수 있으며, 아래 그림에서 소비할 수 있는 첫 번째 데이터를 꺼낼 수 있습니다. redis

동시에 목록의 lpop/ 및 blpop(팝업 시 원래 큐의 데이터가 자동으로 삭제됨) data)와 다르다는 점에 유의하세요.

데이터를 얻었지만 zrem을 사용하지 않는 경우 삭제하려면 이 데이터가 여전히 zset에 존재하기 때문에 다른 사람이 읽을 수 있습니다.

그러나 zrem이 삭제(소비)되는 경우 코드에서 반환 여부도 판단해야 합니다. zrem의 값은 0보다 크며, 이는 이 메시지를 성공적으로 선점했는지 여부와 성공 후 올바르게 소비하는지 여부를 나타냅니다.

코드 구현

Publisher

Subscribers

지연 효과 테스트

전체 코드 주소: https://github.com/qiaomengnan16 /redis-demo/tree/main/redis-delayed-queue

최적화, lua를 사용하여 구현

위에 구현한 지연 큐에 문제가 있습니다. 즉, zrem을 사용하여 데이터를 계속해서 읽게 되면 여러 라운드 동안 잡지 못하고 리소스가 낭비될 수 있으므로 Lua 스크립트를 통해 최적화할 수 있습니다. zrangebyscore 및 zrem을 원자적 작업으로 설정하면 다중 스레드 경합을 방지하고 확보할 수 없는 리소스 낭비를 피할 수 있습니다.

결론

RabbitMQ와 같은 일부 전문 큐 미들웨어는 적용하기가 더 복잡하고 운영 및 유지 관리 비용이 증가합니다. 메시지를 보내기 전에 Exchange 스위치를 만든 다음 Exchange 스위치를 만들어야 합니다. Queue, 그리고 Exchange와 Queue를 바인딩하려면 Exchange와 일치하도록 메시지를 보낼 때 라우팅 키를 지정해야 하며 최종적으로 Queue에 도달해야 합니다.

시나리오가 간단하다면 redis를 사용하여 대기열을 구현할 수 있지만 그렇게 해야 합니다. Redis에는 전문 대기열의 특성이 없습니다. 이는 메시지를 신뢰할 수 없음을 의미합니다. 100% 신뢰성이 필요한 경우에도 메시지가 사라집니다. 전문 대기열 미들웨어 및 ack와 같은 기타 메커니즘을 보장합니다.

더 많은 프로그래밍 관련 지식을 보려면

프로그래밍 소개

를 방문하세요! !

위 내용은 Redis의 메시지 큐 및 지연된 메시지 큐 구현 방법에 대한 간략한 설명의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!