<iframe data-google-container-id="a!2" data-google-query-id="CMm_uMuAlP0CFQnsvAodEP4CUw" data-load-complete="true" frameborder="0" height="280" id="aswift_1" marginheight="0" marginwidth="0" name="aswift_1" scrolling="no" src="https://googleads.g.doubleclick.net/pagead/ads?client=ca-pub-6384567588307613&output=html&h=280&slotname=3921475131&adk=1931977407&adf=2062698613&pi=t.ma~as.3921475131&w=820&fwrn=4&fwrnh=100&lmt=1676343073&rafmt=1&format=820x280&url=https%3A%2F%2Fwww.jb51.net%2Farticle%2F230977.htm&fwr=0&rpe=1&resp_fmts=3&wgl=1&uach=WyJXaW5kb3dzIiwiMC4zLjAiLCJ4ODYiLCIiLCIxMDguMC4xNDYyLjU0IixbXSxmYWxzZSxudWxsLCI2NCIsW1siTm90P0FfQnJhbmQiLCI4LjAuMC4wIl0sWyJDaHJvbWl1bSIsIjEwOC4wLjUzNTkuMTI1Il0sWyJNaWNyb3NvZnQgRWRnZSIsIjEwOC4wLjE0NjIuNTQiXV0sZmFsc2Vd&dt=1676343072420&bpp=3&bdt=645&idt=218&shv=r20230209&mjsv=m202302080101&ptt=9&saldr=aa&abxe=1&cookie=ID%3D3f6ee80e2a3f905b-2219d19c9ad900fc%3AT%3D1676008321%3ART%3D1676008321%3AS%3DALNI_MbH1MqxHmDku8NHi6Itz8cKksnkEw&gpic=UID%3D00000bbf5e3d2a8d%3AT%3D1676008321%3ART%3D1676343010%3AS%3DALNI_MaX_rmRu6Nvh1E9VdWIbK22__-vRA&prev_fmts=0x0&nras=1&correlator=4655122113518&frm=20&pv=1&ga_vid=1071489551.1676343073&ga_sid=1676343073&ga_hid=904200082&ga_fc=0&u_tz=480&u_his=1&u_h=864&u_w=1536&u_ah=826&u_aw=1536&u_cd=24&u_sd=1.25&dmc=8&adx=175&ady=908&biw=1519&bih=755&scr_x=0&scr_y=3437&eid=44759926%2C44759842%2C44759875%2C31072259&oid=2&pvsid=3959018150677192&tmod=607870355&uas=0&nvt=2&ref=https%3A%2F%2Fwww.baidu.com%2Flink%3Furl%3D5fOkR6m6rE-JZ1iisTTQ3OnDb0gwnr9-xA9b4gJNwDyIgWwcfZDB3z1o2IzkPEu4QdFMR2G-6ZEKdgbfY9apaq%26wd%3D%26eqid%3Df7b71537000080be0000000663eaf69a&eae=0&fc=1920&brdim=0%2C0%2C0%2C0%2C1536%2C0%2C1536%2C826%2C1536%2C755&vis=1&rsz=%7C%7CeE%7C&abl=CS&pfx=0&fu=128&bc=31&ifi=2&uci=a!2&fsb=1&xpc=lXSVYO8GkE&p=https%3A//www.jb51.net&dtd=609" width="820"></iframe>
+
目录
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阻塞式处理消息队列的详细内容，更多关于Java redisTemplate 处理消息队列的资料请关注脚本之家其它相关文章！
标签：return,increment,队列,key,Java,public,redisTemplate From： https://www.cnblogs.com/roak/p/17118914.html