Java redisTemplate阻塞式处理消息队列
用redis中的List可以实现队列,这样可以用来做消息处理和任务调度的队列。因此,本文将主要为大家介绍如何利用redisTemplate处理消息队列,感兴趣的小伙伴可以了解一下
目录
Redis 消息队列
redis五种数据结构
队列生产者
队列消费者
测试类
并发情况下使用increment递增
补充
Redis 消息队列
redis五种数据结构
队列生产者
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 | 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(); } } } } |
队列消费者
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 | 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 + "处理成功,被清除" ); } } } } |
测试类
1 2 3 4 5 6 7 8 9 10 | @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递增
线程一:
1 2 | Long increment = redisTemplate.opsForValue().increment( "increment" , 1L); log.info( "队列消费者 >> increment递增: {}" , increment); |
线程二:
1 2 | Long increment = redisTemplate.opsForValue().increment( "increment" , 1L); log.info( "生产者队列 >> increment递增: {}" , increment); |
补充
redisTemplate处理/获取redis消息队列
(参考代码)
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 | /** * 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 ; } } } |
以上就是Java redisTemplate阻塞式处理消息队列的详细内容
原文链接:https://blog.csdn.net/o_o814222198/article/details/121643919