JavaredisTemplate 阻塞式处理消息队列

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>
* 队列生产者
* </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);
  }

并发情况下使用increment递增

线程一：

Long increment = redisTemplate.opsForValue().increment("increment", 1L);
          log.info("队列消费者 >> increment递增: {}", increment);

线程二：

Long increment = redisTemplate.opsForValue().increment("increment", 1L);
          log.info("生产者队列 >> increment递增: {}", increment);

补充

redisTemplate处理/获取redis消息队列

（参考代码）

/**
* 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 处理消息队列的资料请关注编程宝库其它相关文章！

可见性要实现共享变量的可见性，必须保证两点：线程修改后的共享变量值能够及时从工作内存刷新到主内存中其他线程能够及时把共享变量的最新值从主内存更新到自己的工作内存中Java语言层面支持的可见性的 ...