Example Analysis of Java redisTemplate blocking message queue

This article mainly introduces the example analysis of Java redisTemplate blocking message queue, which is very detailed and has certain reference value. Friends who are interested must finish it!

Redis message queuing

Five data structures of redis

Queue producer 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;/** *

* queue producer *

* * @ SINCE 21:03 on 2021-11-30 * @ AUTHOR dispark * @ Date: 2021-11-30 21:03 * / @ Slf4jpublic class QueueProducer implements Runnable {/ * producer queue key * / public static final String QUEUE_PRODUCTER = "queue-producter"; private RedisTemplate redisTemplate; public QueueProducer (RedisTemplate redisTemplate) {this.redisTemplate = redisTemplate;} @ Override public void run () {Random random = new Random () While (true) {try {Thread.sleep (random.nextInt (600) + 600); The simulation generates a task UUID queueProducerId = UUID.randomUUID (); / / 2. Insert the task into the task queue: queue-producter redisTemplate.opsForList (). LeftPush (QUEUE_PRODUCTER, queueProducerId.toString ()); log.info ("produce a piece of data > {}", queueProducerId.toString ());} catch (Exception e) {e.printStackTrace () } queue Consumer package cn.stylefeng.guns.knowledge.modular.knowledge.schedule;import lombok.extern.slf4j.Slf4j;import org.springframework.data.redis.core.RedisTemplate;import java.util.Random;/** *

* queue consumers *

* * @ SINCE 21:14 on 2021-11-30 * @ 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 redisTemplate; public QueueConsumer (RedisTemplate redisTemplate) {this.redisTemplate = redisTemplate } / * * feature description: value-- it is recommended to use * * @ author dispark * @ date 21:17 on 2021-11-30 * / @ Override public void run () {Random random = new Random (); while (true) {/ / 1. Get a task from the task queue "queue-producter" and put the task in the staging queue "tmp-queue" Long ququeConsumerId = redisTemplate.opsForList (). RightPush (QUEUE_PRODUCTER, TMP_QUEUE); / / 2. Processing tasks-purely business logic, simulate: sleep try {Thread.sleep (1000);} catch (InterruptedException e) {e.printStackTrace ();} / / 3. Simulate the accidental phenomena of success and failure, and simulate the situation of failure. The probability is 2 to 13 if (random.nextInt (13)% 7 = = 0) {/ / 4. Bounce back the task queue "queue-producter" redisTemplate.opsForList (). RightPush (TMP_QUEUE, QUEUE_PRODUCTER); log.info (ququeConsumerId + "failed processing, bounced back to the task queue") from the staging queue "tmp-queue";} else {/ / 5. If the simulation is successful, clear this task from the staging queue "tmp-queue" redisTemplate.opsForList (). RightPop (TMP_QUEUE); log.info (ququeConsumerId + "processed successfully, cleared");} test class @ Test public void QueueThreadTotalEntry () throws Exception {/ / 1. Start a producer thread to simulate the generation of tasks new Thread (new QueueProducer (redisTemplate)). Start (); Thread.sleep (15000); / / 2. Start a thread to simulate task processing new Thread (new QueueConsumer (redisTemplate)). Start (); / / 3. Main thread Thread.sleep (Long.MAX_VALUE);} use increment increment in concurrency

Thread one:

Long increment = redisTemplate.opsForValue () .increment ("increment", 1L); log.info ("queue consumer > > increment increment: {}", increment)

Thread two:

Long increment = redisTemplate.opsForValue () .increment ("increment", 1L); log.info ("producer queue > > increment increment: {}", increment)

Supplement

RedisTemplate processes / acquires redis message queues

(reference code)

/ * redis message queuing * / @ Componentpublic class RedisQueue {@ Autowired private RedisTemplate redisTemplate / * *-- redis message queue-- * / * stored value * @ param key key * @ param value value * @ Return * / public boolean lpush (String key Object value) {try {redisTemplate.opsForList () .leftPush (key, value) Return true;} catch (Exception e) {e.printStackTrace (); return false;}} / * * values-* @ param key key * @ return * / public Object rpop (String key) {try {return redisTemplate.opsForList () .rightPop (key) } catch (Exception e) {e.printStackTrace (); return null }} / * * value-it is recommended to use the * @ param key key * @ param timeout timeout * @ param timeUnit given cell granularity time period * TimeUnit.DAYS / day * TimeUnit.HOURS / / hour * TimeUnit.MINUTES / / minute * TimeUnit.SECONDS / / second * TimeUnit.MILLISECONDS / / millisecond * @ 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 }} / * View values * @ param key key * @ param start start * @ param end ends 0 to-1 represents all values * @ return * / public List lrange (String key, long start, long end) {try {return redisTemplate.opsForList () .range (key, start, end) } catch (Exception e) {e.printStackTrace (); return null;} these are all the contents of the article "sample Analysis of Java redisTemplate blocking message queues". Thank you for reading! Hope to share the content to help you, more related knowledge, welcome to follow the industry information channel!