SpringBoot集成Redis实现消息队列的方法

更新时间：2021年02月10日 09:31:20 作者：新猿一马

这篇文章主要介绍了SpringBoot集成Redis实现消息队列的方法，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧

list 原理说明

Redis 的 list 是按照插入顺序排序的字符串链表。

如图所示，可以通过 lpush 和 rpop 或者 rpush 和 lpop 实现消息队列。

1 lpush 和 rpop

2 rpush 和 lpop

消息队列功能实现

引入 Redis 依赖

<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>

applicat.yml添加Redis配置

spring:
 redis:
  host: 127.0.0.1
  database: 0
  port: 6379
  jedis:
   pool:
    max-active: 256
    max-idle: 8
    min-idle: 1

Redis配置类

package com.sb.config;
 
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;
 
@Configuration
public class RedisConfig {
 
  @Autowired
  private RedisConnectionFactory redisConnectionFactory;
 
  @Bean
  public RedisTemplate<String, Object> redisTemplate() {
    RedisTemplate<String, Object> template = new RedisTemplate<>();
    template.setConnectionFactory(redisConnectionFactory);
    template.setKeySerializer(new StringRedisSerializer());
    template.setValueSerializer(new StringRedisSerializer());
    template.afterPropertiesSet();
    return template;
  }
 
}

MQ发送和接收接口

package com.sb.service;
 
public interface MQService {
 
  void produce(String string);
 
  void consume();
}

MQ发送和接收实现类

package com.sb.service.impl;
 
import com.sb.service.MQService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.lang.Nullable;
import org.springframework.stereotype.Service;
 
import javax.annotation.Resource;
import java.util.List;
 
@Service
public class MQServiceImpl implements MQService {
 
  private static Logger log = LoggerFactory.getLogger(MQServiceImpl.class);
 
  private static final String MESSAGE_KEY = "message:queue";
 
  @Resource
  private RedisTemplate redisTemplate;
 
  @Override
  public void produce(String string) {
    redisTemplate.opsForList().leftPush(MESSAGE_KEY, string);
  }
 
  @Override
  public void consume() {
    String string = (String) redisTemplate.opsForList().rightPop(MESSAGE_KEY);
    log.info("consume : {}", string);
  }
 
}

MQ发送和接收API接口

package com.sb.controller;
 
import com.sb.service.MQService;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
 
import javax.annotation.Resource;
 
@RestController
@RequestMapping(value="/api")
public class MQController {
 
  @Resource
  private MQService mQService;
 
  @RequestMapping(value = "/produce", method=RequestMethod.GET)
  public void produce(@RequestParam(name = "key") String key) {
    mQService.produce(key);
  }
 
  @RequestMapping(value="/consume", method=RequestMethod.GET)
  public void consume() {
    while (true) {
      mQService.consume();
    }
  }
 
}

消息队列功能测试

调用 http://localhost:8080/api/produce 接口往队列里面添加 a、b、c、d元素。

调用 http://localhost:8080/api/consume 消费队列里面的元素。

从截图我们可以看到，即使当队列为空，消费者依然在不停的 pop 数据，这就是浪费生命的空轮询。

那如何解决这个空轮询的问题呢？

你也许会想使用 Thread.sleep() 让消费者线程隔一段时间再消费。

使用 Thread.sleep() 会有什么问题么？

A 如果生产者速度大于消费者消费速度，消息队列长度会一直增大，时间久了会占用大量内存空间。

B 如果睡眠时间过长，这样不能处理一些时效性的消息，睡眠时间过短，也会在连接上造成比较大的开销。

有没有更优雅和更合适的方式呢？

brpop 和 blpop 实现阻塞读取，下面以 blpop 为例来说明问题。

blpop 理论说明

blpop 命令

blpop key1...keyN timeout

blpop 说明

blpop 是阻塞式列表的弹出原语。当给定列表内没有任何元素可供弹出的时候，连接将被 blpop 命令阻塞。直到有另一个客户端对给定的这些 key 的任意一个执行 lpush 或 rpush 命令为止。

当给定多个 key 参数时，按参数 key 的先后顺序依次检查各个列表，弹出第一个非空列表的头元素。

key1...keyN：表示不同的队列名。

timeout：阻塞队列超时时间。

blpop 代码实现

public void blockingConsume() {
  List<Object> obj = redisTemplate.executePipelined(new RedisCallback<Object>() {
    @Nullable
    @Override
    public Object doInRedis(RedisConnection connection) throws DataAccessException {
      //队列没有元素会阻塞操作，直到队列获取新的元素或超时
      return connection.bLPop(TIME_OUT, MESSAGE_KEY.getBytes());
    }
  },new StringRedisSerializer());
 
  for (Object str: obj) {
    log.info("blockingConsume : {}", str);
  }
}

阻塞线程每隔10s超时执行一次。该方法解决了 CPU 空转的问题。

到此这篇关于SpringBoot集成Redis实现消息队列的方法的文章就介绍到这了,更多相关SpringBoot Redis消息队列内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:

浅谈java的守护线程与非守护线程
这篇文章主要介绍了浅谈java的守护线程与非守护线程，小编觉得挺不错的，现在分享给大家，也给大家做个参考。一起跟随小编过来看看吧
2017-10-10
java循环结构、数组的使用小结
这篇文章主要介绍了java循环结构、数组的使用小结，本文通过实例代码给大家介绍的非常详细，对大家的学习或工作具有一定的参考借鉴价值,需要的朋友可以参考下
2021-09-09
SpringMvc微信支付回调示例代码
微信一直是一个比较热门的词汇，今天这篇文章主要介绍的是SpringMvc微信支付回调的示例代码，对大家开发微信支付具有一定的参考借鉴价值，下面来一起看看吧。
2016-09-09
Springboot实例讲解实现宠物医院管理系统流程
读万卷书不如行万里路，只学书上的理论是远远不够的，只有在实战中才能获得能力的提升，本篇文章手把手带你用Springboot实现宠物医院综合管理系统,大家可以在过程中查缺补漏，提升水平
2022-06-06
Java 随机取字符串的工具类
随机数在实际中使用很广泛，比如要随即生成一个固定长度的字符串、数字。或者随即生成一个不定长度的数字、或者进行一个模拟的随机选择等等。Java提供了最基本的工具，可以帮助开发者来实现这一切
2014-01-01
springboot操作静态资源文件的方法
这篇文章主要介绍了springboot操作静态资源文件的方法，本文给大家提到了两种方法，小编在这里比较推荐第一种方法，具体内容详情大家跟随脚本之家小编一起看看吧
2018-07-07
聊一聊Java中的Steam流
当我们需要处理的数据量很大的时候，为了提高性能，就需要使用到并行处理，这样的处理方式是很复杂的，流可以帮助开发者节约宝贵的时间，让以上的事情变得轻松，本文就和大家聊一聊Java中的Steam流，感兴趣的同学跟着小编一起来看看吧
2023-07-07
关于分布式锁(Redisson)的原理分析
这篇文章主要介绍了关于分布式锁(Redisson)的原理，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
2022-08-08
SpringBoot事件发布与监听超详细讲解
今天去官网查看spring boot资料时，在特性中看见了系统的事件及监听章节,所以下面这篇文章主要给大家介绍了关于SpringBoot事件发布和监听的相关资料,文中通过实例代码介绍的非常详细,需要的朋友可以参考下
2022-11-11
深入了解SpringBoot中@InitBinder注解的使用
这篇文章主要介绍了深入了解SpringBoot中@InitBinder注解的使用,@InitBinder注解可以作用在被@Controller注解的类的方法上,表示为当前控制器注册一个属性编辑器,用于对WebDataBinder进行初始化,且只对当前的Controller有效,需要的朋友可以参考下
2023-10-10