Redis 큐 우선순위 코드 예시 구현 방법에 대한 자세한 설명

Redis를 사용하여 메시지 큐를 만드는 방법

우선 Redis는 캐싱용으로 설계되었지만, 그 자체의 특성상 메시지 큐에도 사용할 수 있습니다. 여기에는 사용할 수 있는 여러 차단 API가 있으며 메시지 대기열을 수행할 수 있는 기능을 제공하는 것이 바로 이러한 차단 API입니다.

"데이터베이스가 모든 문제를 해결한다"는 생각 아래 메시지 대기열을 사용하지 않고도 요구 사항을 완료할 수 있다고 상상해 보세요. 모든 작업을 데이터베이스에 저장한 후 지속적인 폴링을 통해 처리합니다. 이 접근 방식으로 작업을 완료할 수는 있지만 매우 조잡합니다. 그러나 데이터베이스 인터페이스가 차단 방법을 제공하는 경우 폴링 작업을 피할 수 있습니다. 데이터베이스를 메시지 대기열로 사용할 수도 있지만 현재 데이터베이스에는 그러한 인터페이스가 없습니다.

또한 FIFO와 같은 메시지 대기열의 다른 기능도 쉽게 구현할 수 있습니다. 처음부터 데이터를 가져오고 끝 부분에서 데이터를 채우려면 List 개체만 있으면 됩니다.

Redis는 목록 개체 blpop brpop 인터페이스와 Pub/Sub(게시/구독)의 일부 인터페이스 덕분에 메시지 대기열을 수행할 수 있습니다. 모두 차단 버전이므로 메시지 대기열로 사용할 수 있습니다.

rabbitmq의 우선순위접근 방식

현재 많은 성숙한 메시지 대기열 제품이 있으며 그 중 유명한 제품은 Rabbitmq입니다. 사용하기가 비교적 간단하고 기능이 비교적 풍부하여 일반적인 상황에서는 충분합니다. 하지만 매우 짜증나는 점 중 하나는 우선순위를 지원하지 않는다는 것입니다.

예를 들어, 이메일을 보내는 작업에서 일부 권한 있는 사용자는 자신의 이메일이 더 적시에 전송되기를 원하거나 최소한 일반 사용자보다 우선순위를 부여하기를 바랍니다. 기본적으로 Rabbitmq는 이를 처리할 수 없습니다. Rabbitmq에 던져진 작업은 FIFO 선입선출입니다. 그러나 이러한 우선순위를 지원하기 위해 몇 가지 해결 방법을 사용할 수 있습니다. 여러 대기열을 생성하고 Rabbitmq 소비자에 대한 해당 라우팅 규칙을 설정합니다.

예를 들어 기본적으로 이런 대기열이 있습니다. 목록을 사용하여 [task1, task2, task3]을 시뮬레이션하고 소비자는 FIFO 원칙에 따라 하나씩 작업을 꺼내서 처리합니다. 우선순위가 높은 작업이 들어오면 마지막[task1, task2, task3, higitask1]만 처리할 수 있습니다. 하지만 두 개의 대기열을 사용하는 경우에는 높은 우선순위 대기열과 일반 우선순위 대기열이 있습니다. 보통 우선순위[task1, task2, task3], 높은 우선순위[hightask1] 그런 다음 소비자의 라우팅을 설정하고 소비자가 임의의 대기열에서 데이터를 무작위로 가져오도록 합니다.

그리고 우선 순위가 높은 대기열을 특별히 처리하는 소비자를 정의할 수 있습니다. 소비자는 유휴 상태일 때 우선 순위가 낮은 대기열의 데이터를 처리하지 않습니다. 이는 은행의 VIP 카운터와 비슷하며, 일반 고객들이 은행에서 번호를 받기 위해 줄을 서고 있는데, VIP가 오면 일반 회원들 앞에 있는 번호표를 뽑지는 않지만, VIP 채널로 더 빠르게 직접 이동할 수 있습니다.

rabbitmq를 사용하여 우선순위 메시지 대기열을 지원하는 경우 위에서 언급한 동일한 은행의 VIP 회원과 마찬가지로 다른 채널을 거쳐야 합니다. 그러나 이 방법은 상대적인 우선순위일 뿐 절대적인 우선순위 제어를 달성할 수 없습니다. 예를 들어, 위의 솔루션은 다른 일반 작업보다 우선순위가 높은 특정 작업이 절대적인 의미에서 먼저 처리되기를 바랍니다. 일하다. . Rabbitmq의 소비자는 비어 있을 때 관심을 갖는 대기열에서 대기열의 첫 번째 데이터를 "무작위로" 처리하는 방법만 알고 있기 때문에 어떤 대기열을 먼저 가져갈지 제어할 수 없습니다. 또는 보다 세분화된 우선순위 제어. 또는 시스템에 10개 이상의 우선순위가 설정되어 있습니다. 이런 식으로 Rabbitmq를 사용하는 것도 어렵습니다.

하지만 Redis를 대기열로 사용하면 위의 요구 사항을 충족할 수 있습니다.

메시지 대기열이 필요한 이유

시스템에 메시지 대기열 메커니즘을 도입하는 것은 시스템에 매우 큰 개선입니다. 예를 들어, 웹 시스템에서는 사용자가 특정 작업을 수행한 후 이메일 알림을 사용자의 사서함으로 보내야 합니다. 동기식 방법을 사용하여 사용자가 이메일 전송 및 완료 후 사용자에게 피드백을 기다리게 할 수 있지만 이로 인해 네트워크 불확실성으로 인해 사용자가 오랜 시간 기다리게 되어 사용자 경험에 영향을 미칠 수 있습니다. .

일부 시나리오에서는 동기 방식을 사용하여 완료를 기다리는 것이 불가능하며 이러한 작업에는 백그라운드에서 많은 시간이 필요합니다. 예를 들어 극단적인 예로 온라인 컴파일 시스템 작업의 경우 백그라운드 컴파일을 완료하는 데 30분이 걸립니다. 이 시나리오의 설계에서는 동기적으로 기다린 다음 피드백을 제공하는 것이 불가능합니다. 먼저 사용자에게 피드백한 다음 비동기 처리가 완료된 다음 처리가 완료될 때까지 기다린 후 지침에 따라 사용자에게 피드백을 제공해야 합니다. 상황.

또한 메시지 대기열은 시스템 처리 용량이 제한된 상황에 적합합니다. 대기열 메커니즘은 먼저 작업을 임시로 저장하는 데 사용되며 시스템은 대기열에 있는 작업을 하나씩 처리합니다. 이러한 방식으로 시스템 처리량이 부족한 경우에도 동시성이 높은 작업을 안정적으로 처리할 수 있습니다.

메시지 대기열은 대기열 메커니즘으로 사용될 수 있습니다. 시스템이 대기열 메커니즘을 사용해야 하는 한 메시지 대기열을 사용할 수 있습니다.

redis 메시지 대기열 우선순위 구현

몇 가지 기본 Redis 기본 사항에 대한 설명

redis> blpop tasklist 0
"im task 01"

이 예제에서는 blpop 명령을 사용하여 작업 목록 목록에서 첫 번째 데이터를 차단 방식으로 가져오고 마지막 매개변수는 대기 시간 초과입니다. 0으로 설정하면 무기한 대기한다는 의미입니다. 또한 Redis에 저장된 데이터는 string 유형만 가능하므로 작업을 전송할 때는 string만 전달할 수 있습니다. 해당 데이터를 json 형식의 문자열로 간단히 직렬화한 다음 소비자 측에서 변환하면 됩니다.

여기서 예제 언어는 Python을 사용하고, redis에 연결된 라이브러리는 redis-py를 사용합니다. 프로그래밍 기술이 있다면 원하는 언어로 전환해도 문제가 없습니다.

1. 간단한 FIFO 대기열

import redis, time
def handle(task):
    print task
    time.sleep(4)
 
def main():
    pool = redis.ConnectionPool(host='localhost', port=6379, db=0)
    r = redis.Redis(connection_pool=pool)
    while 1:
        result = r.brpop('tasklist', 0)
        handle(result[1])
 
if name == "main":
    main()

위의 예는 가장 간단한 소비자이기도 합니다. 우리는 무한 루프를 통해 Redis 대기열에서 지속적으로 데이터를 가져옵니다. 큐에 데이터가 없으면 타임아웃 없이 그곳에서 차단됩니다. 데이터가 있으면 꺼내서 실행합니다.

일반적으로 복잡한 문자열을 꺼내서 형식을 지정하고 처리 함수에 전달해야 할 수도 있지만 단순화를 위해 예제에서는 일반 문자열을 사용합니다. 또한 예제의 처리 기능은 어떠한 처리도 수행하지 않으며 시간이 많이 소요되는 작업을 시뮬레이션하기 위해 절전 모드에만 사용됩니다.

생산자를 시뮬레이션하기 위해 또 다른 Redis 클라이언트를 열었습니다. 내장 클라이언트로 충분합니다. 작업 목록 대기열에 더 많은 데이터를 넣습니다.

redis> lpush tasklist 'im task 01'
redis> lpush tasklist 'im task 02'
redis> lpush tasklist 'im task 03'
redis> lpush tasklist 'im task 04'
redis> lpush tasklist 'im task 05'

그러면 소비자 측에서는 이러한 시뮬레이션된 작업이 하나씩 소비되는 것을 볼 수 있습니다.

2. 단순 우선순위 큐

우선순위가 낮은 작업보다 우선순위가 높은 작업만 먼저 처리해야 하는 간단한 요구사항을 가정합니다. 다른 작업의 순서는 중요하지 않습니다. 이 경우 우선순위가 높은 작업을 만났을 때 대기열 뒤로 밀어넣는 대신 대기열 앞쪽으로 밀어넣기만 하면 됩니다.

우리 대기열은 Redis 목록을 사용하기 때문에 구현하기 쉽습니다. 높은 우선순위를 만나면 rpush를 사용하세요. 그러면 낮은 우선순위를 만나면 항상 높은 우선순위가 낮은 우선순위보다 먼저 실행되는 것을 볼 수 있습니다. 그러나 이 솔루션의 단점은 우선 순위가 높은 작업의 실행 순서가 선입후출이라는 것입니다.

3. 더욱 완전한 대기열

예제 2에서는 우선 순위가 높은 작업이 대기열 맨 앞에 배치되고, 우선 순위가 낮은 작업이 대기열 끝에 배치됩니다. 이는 우선순위가 높은 작업 간의 순서를 보장하지 않습니다.

모든 작업의 ​​우선순위가 높을 때 실행 순서가 뒤바뀐다고 가정해 보세요. 이는 분명히 대기열의 FIFO 원칙을 위반합니다.

하지만 조금만 개선하면 우리 대기열도 개선될 수 있습니다.

rabbitmq를 사용하는 것과 유사하게 우선순위가 높은 대기열과 낮은 우선순위 대기열이라는 두 개의 대기열을 설정합니다. 우선순위가 높은 작업은 높은 우선순위 대기열에 배치되고, 우선순위가 낮은 작업은 낮은 우선순위 대기열에 배치됩니다. redis와 Rabbitmq의 차이점은 대기열 소비자에게 어느 대기열에서 먼저 읽도록 요청할 수 있다는 것입니다.

redis> lpush tasklist 'im task 01'
redis> lpush tasklist 'im task 02'
redis> rpush tasklist 'im high task 01'
redis> rpush tasklist 'im high task 01'
redis> lpush tasklist 'im task 03'
redis> rpush tasklist 'im high task 03'

위 코드는 'high_task_queue' 및 'low_task_queue' 두 대기열에서 차단 방식으로 데이터를 가져옵니다. 첫 번째 대기열을 사용할 수 없으면 두 번째 대기열에서 데이터를 가져옵니다.

따라서 목표를 달성하려면 대기열 소비자에게 이러한 개선만 하면 됩니다.

def main():
    pool = redis.ConnectionPool(host='localhost', port=6379, db=0)
    r = redis.Redis(connection_pool=pool)
    while 1:
        result = r.brpop(['high_task_queue', 'low_task_queue'], 0)
        handle(result[1])

위의 테스트를 통해 높은 우선순위가 먼저 실행되고, 높은 우선순위 사이에서도 FIFO 원칙이 보장되는 것을 알 수 있습니다.

이 솔루션을 사용하면 높음, 중간, 낮음 등 세 가지 수준의 다양한 단계에서 우선순위 대기열을 지원할 수 있습니다.

4. 우선순위 수준이 많은 상황

이런 수요가 있고 우선순위가 높음, 중간, 낮음 또는 0~10의 단순한 고정 수준이 아니라고 가정해 보겠습니다. 하지만 0-99999와 같은 레벨이 너무 많습니다. 그러면 세 번째 옵션은 적합하지 않습니다.

redis에는 정렬 세트처럼 정렬 가능한

이 있지만 차단 버전의 인터페이스가 없는 것이 아쉽습니다. 따라서 다른 방법을 통해 목적을 달성하기 위해 목록 유형만 사용할 수 있습니다. 간단한 방법은 하나의 대기열만 설정하고 우선순위에 따라 정렬되도록 하는 것입니다. 그런 다음

방법을 사용하여 작업에 적합한 위치를 찾고 lset 명령을 통해 해당 위치에 삽입합니다. 예를 들어 대기열에는 쓰기 우선순위가 [1, 3, 6, 8, 9, 14]인 작업이 포함되어 있습니다. 우선순위가 7인 작업이 오면 자체 바이너리 알고리즘을 사용하여 대기열에서 데이터를 하나씩 가져옵니다. 대상 데이터와 비교하여 해당 위치를 계산한 후 지정된 위치에 삽입합니다.

바이너리 검색은 상대적으로 빠르며, redis 자체도 메모리에 있기 때문에 이론적으로는 속도를 보장할 수 있습니다. 하지만 데이터의 양이 정말 크다면 어떤 방식으로든 조정할 수도 있습니다.

세 번째 옵션을 떠올려 보세요. 세 번째 옵션을 결합하면 오버헤드가 크게 줄어듭니다. 예를 들어, 데이터 볼륨이 100,000인 대기열의 경우 우선 순위도 0~100,000 범위에서 무작위로 지정됩니다. 10개 또는 100개의 서로 다른 대기열을 설정할 수 있습니다. 0에서 10,000까지의 우선순위 작업은 대기열 1에 배치되고 10,000에서 20,000까지의 작업은 대기열 2에 배치됩니다. 이러한 방식으로 큐가 여러 레벨로 분할된 후 단일 큐의 데이터가 많이 줄어들므로 이진 검색 매칭의 효율성이 높아집니다. 그러나 데이터가 차지하는 리소스는 기본적으로 변경되지 않습니다. 10만 개의 데이터는 동일한 양의 메모리를 차지해야 합니다. 시스템에 더 많은 대기열이 있다는 것입니다.

위 내용은 Redis 큐 우선순위 코드 예시 구현 방법에 대한 자세한 설명의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!