Java redisTemplate阻塞式处理消息队列

目录

redis 消息队列

redis五种数据结构

队列生产者

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> * 队列生产者国庆阅兵观后感600字 * </p> * * @since 2021/11/30 21:03 * @author dispark * @date: 2021/11/30 21:03 */@slf4jpublic 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 */@slf4jpublic 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 + "处理失败，被弹回任务队列");            } el {                // 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);    }

并发情况下使用increment递增

线程一：

long increment = redistemplate.opsforvalue().increment("increment", 1l);   un前缀的单词         log.info("队列消费者 >> increment递增: {}", increment);

线程二：

long increment = redistemplate.opsforvalue().increment("increment", 1l);            log.info("生产者队列 >> increment递增: {}", increment);

补充

redistemplate处理/获取redis消息队列

（参考代码）

/** * redis消息队列 */@componentpublic 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 假如时光倒流fal;        }    }     /**     * 取值 - <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.conds       //秒     *                 timeunit.milliconds  //毫秒     * @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;        }    } }

以上就是java redistemplate阻塞式处理消息队列的详细内容，更多关于java redistemplate 处理消息队列的资料请关注www.887551.com其它相关文章！