Как Redis реализует очередь сообщений

Redis

Повторное собеседование

В интервью я спросил: если я хочу реализовать очередь в Redis, какую команду я буду использовать? согласно сруководство RedisЧто касается основных операций команды Redis list (Список), существует четыре неблокирующие версии lpush/lpop/rpush/rpop и две блокирующие версии blpop/brpop, где l и r обозначают левое (левое) и правое (правое) аббревиатуры, а push означает добавление. Это называется push, а pop означает pop. Базовая очередь завершается, в основном, удовлетворяя эффект «первым поступил – первым обслужен» (FIFO) Этот экспериментальный репозиторий кода GitHub.

Имитация очереди с командами

RDM Redis Console
连接中...
已连接。
localhost:0>lpush mylist a b c d
"4"
localhost:0>rpop mylist
"a"
localhost:0>rpop mylist
"b"
localhost:0>rpop mylist
"c"
localhost:0>rpop mylist
"d"
localhost:0>rpop mylist
null
localhost:0>

2. Java имитирует очередь для реализации модели производитель-потребитель.

константы конфигурации Redis

package io.charles;

public class Constant {
    /**
     * redis链接地址
     */
    public static final String host = "127.0.0.1";
    /**
     * redis启动端口
     */
    public static final int port = 6379;
    /**
     * 正式队列列表名称
     */
    public static final String task_queue = "task-queue";
    /**
     * 临时队列列表名称
     */
    public static final String tmp_queue = "tmp-queue";
}

TaskProducer имитирует производителя

package io.charles;

/**
 * 用于模拟生产者
 */

import java.util.Random;
import java.util.UUID;

import redis.clients.jedis.Jedis;

public class TaskProducer implements Runnable {
    Jedis jedis = new Jedis(Constant.host, Constant.port);

    public void run() {
        Random random = new Random();
        while (true) {
            try {
                Thread.sleep(random.nextInt(600) + 600);
                // 模拟生成一个任务
                UUID taskid = UUID.randomUUID();
                //将任务插入任务队列：task-queue
                jedis.lpush(Constant.task_queue, taskid.toString());
                System.out.println("插入了一个新的任务： " + taskid);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}

TaskConsumer имитирует потребителя

package io.charles;

import java.util.Random;

import redis.clients.jedis.Jedis;

/**
 * 模拟消费者
 */
public class TaskConsumer implements Runnable {
    Jedis jedis = new Jedis(Constant.host, Constant.port);

    public void run() {
        Random random = new Random();

        while (true) {

            //从任务队列"task-queue"中获取一个任务，并将该任务放入临时队列"tmp-queue"
            String taskid = jedis.rpoplpush(Constant.task_queue, Constant.tmp_queue);

            // 模拟一下：睡觉
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            //模拟成功和失败的偶然现象
            if (random.nextInt(13) % 7 == 0) {// 模拟失败的情况,概率为2/13
                //将本次处理失败的任务从临时队列"tmp-queue"中，弹回任务队列"task-queue"
                jedis.rpoplpush(Constant.task_queue, Constant.tmp_queue);
                System.out.println(taskid + "处理失败，被弹回任务队列");

            } else {// 模拟成功的情况
                // 将本次任务从临时队列"tmp-queue"中清除
                jedis.rpop(Constant.tmp_queue);
                System.out.println(taskid + "处理成功，被清除");
            }
        }
    }
}

TaskShedulerSystem запускает потоки производителя и потребителя.

package io.charles;

public class TaskShedulerSystem {
    public static void main(String[] args) throws Exception {

        // 启动一个生产者线程，模拟任务的产生
        new Thread(new TaskProducer()).start();

        Thread.sleep(15000);

        //启动一个线程者线程，模拟任务的处理
        new Thread(new TaskConsumer()).start();
    }
}

3. Ссылки

  1. Redis китайское руководство по командам
  2. Репозиторий экспериментального кода GitHub
Категории