Java redisTemplate阻塞式處理消息隊列

package cn.stylefeng.guns.knowledge.modular.knowledge.schedule;

import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.RedisTemplate;

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

/**
 * <p>
 * 隊列生產者
 * </p>
 *
 * @SINCE 2021/11/30 21:03
 * @AUTHOR dispark
 * @Date: 2021/11/30 21:03
 */
@Slf4j
public class QueueProducer implements Runnable {

    /**
     * 生產者隊列 key
     */
    public static final String QUEUE_PRODUCTER = "queue-producter";

    private RedisTemplate<String, Object> redisTemplate;

    public QueueProducer(RedisTemplate<String, Object> redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    @Override
    public void run() {
        Random random = new Random();
        while (true) {
            try {
                Thread.sleep(random.nextInt(600) + 600);
                // 1.模擬生成一個任務
                UUID queueProducerId = UUID.randomUUID();
                // 2.將任務插入任務隊列：queue-producter
                redisTemplate.opsForList().leftPush(QUEUE_PRODUCTER, queueProducerId.toString());
                log.info("生產一條數據 >>> {}", queueProducerId.toString());
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}

package cn.stylefeng.guns.knowledge.modular.knowledge.schedule;

import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.RedisTemplate;

import java.util.Random;

/**
 * <p>
 * 隊列消費者
 * </p>
 *
 * @SINCE 2021/11/30 21:14
 * @AUTHOR dispark
 * @Date: 2021/11/30 21:14
 */
@Slf4j
public class QueueConsumer implements Runnable {
    public static final String QUEUE_PRODUCTER = "queue-producter";
    public static final String TMP_QUEUE = "tmp-queue";

    private RedisTemplate<String, Object> redisTemplate;

    public QueueConsumer(RedisTemplate<String, Object> redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    /**
     * 功能描述: 取值 - <brpop：阻塞式> - 推薦使用
     *
     * @author dispark
     * @date 2021/11/30 21:17
     */
    @Override
    public void run() {
        Random random = new Random();
        while (true) {
            // 1.從任務隊列"queue-producter"中獲取一個任務，並將該任務放入暫存隊列"tmp-queue"
            Long ququeConsumerId = redisTemplate.opsForList().rightPush(QUEUE_PRODUCTER, TMP_QUEUE);
            // 2.處理任務----純屬業務邏輯，模擬一下：睡覺
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            // 3.模擬成功和失敗的偶然現象,模擬失敗的情況,概率為2/13
            if (random.nextInt(13) % 7 == 0) {
                // 4.將本次處理失敗的任務從暫存隊列"tmp-queue"中，彈回任務隊列"queue-producter"
                redisTemplate.opsForList().rightPush(TMP_QUEUE, QUEUE_PRODUCTER);
                log.info(ququeConsumerId + "處理失敗，被彈回任務隊列");
            } else {
                // 5. 模擬成功的情況,將本次任務從暫存隊列"tmp-queue"中清除
                redisTemplate.opsForList().rightPop(TMP_QUEUE);
                log.info(ququeConsumerId + "處理成功，被清除");
            }
        }
    }
}

    @Test
    public void QueueThreadTotalEntry() throws Exception {
        // 1.啟動一個生產者線程，模擬任務的產生
        new Thread(new QueueProducer(redisTemplate)).start();
        Thread.sleep(15000);
        // 2.啟動一個線程者線程，模擬任務的處理
        new Thread(new QueueConsumer(redisTemplate)).start();
        // 3.主線程
        Thread.sleep(Long.MAX_VALUE);
    }

Long increment = redisTemplate.opsForValue().increment("increment", 1L);
            log.info("隊列消費者 >> increment遞增: {}", increment);

Long increment = redisTemplate.opsForValue().increment("increment", 1L);
            log.info("生產者隊列 >> increment遞增: {}", increment);

/**
 * redis消息隊列
 */
@Component
public class RedisQueue {
    @Autowired
    private RedisTemplate redisTemplate;
 
 
    /** ---------------------------------- redis消息隊列 ---------------------------------- */
    /**
     * 存值
     * @param key 鍵
     * @param value 值
     * @return
     */
    public boolean lpush(String key, Object value) {
        try {
            redisTemplate.opsForList().leftPush(key, value);
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }
 
    /**
     * 取值 - <rpop：非阻塞式>
     * @param key 鍵
     * @return
     */
    public Object rpop(String key) {
        try {
            return redisTemplate.opsForList().rightPop(key);
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }
 
    /**
     * 取值 - <brpop：阻塞式> - 推薦使用
     * @param key 鍵
     * @param timeout 超時時間
     * @param timeUnit 給定單元粒度的時間段
     *                 TimeUnit.DAYS          //天
     *                 TimeUnit.HOURS         //小時
     *                 TimeUnit.MINUTES       //分鐘
     *                 TimeUnit.SECONDS       //秒
     *                 TimeUnit.MILLISECONDS  //毫秒
     * @return
     */
    public Object brpop(String key, long timeout, TimeUnit timeUnit) {
        try {
            return redisTemplate.opsForList().rightPop(key, timeout, timeUnit);
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }
 
    /**
     * 查看值
     * @param key 鍵
     * @param start 開始
     * @param end 結束 0 到 -1代表所有值
     * @return
     */
    public List<Object> lrange(String key, long start, long end) {
        try {
            return redisTemplate.opsForList().range(key, start, end);
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }
 
}