SpringBoot 集成 Redis 实现消息队列

青旅半醒 2022-10-29 07:48 197阅读 0赞

![format_png][]

点击上方蓝字关注我们

## list 原理说明 ##

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

![format_png 1][]

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

1 lpush 和 rpop

![format_png 2][]

2 rpush 和 lpop

![format_png 3][]

## 消息队列功能实现 ##

**引入 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元素。

![format_png 4][]

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

![format_png 5][]

从截图我们可以看到，即使当队列为空，消费者依然在不停的 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**：阻塞队列超时时间。

![format_png 6][]

## 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);
        }
    }

![format_png 7][]

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

[![format_png 8][]][format_png 8 1]

[用java做一个能赚钱的微信群聊机器人（PC协议）][format_png 8 1]

[![format_png 9][]][format_png 9 1]

[【案例】有什么副业是可以做的?][format_png 9 1]

[![format_png 10][]][format_png 10 1]

[ZooKeeper典型应用场景][format_png 10 1]

[![format_png 11][]][format_png 11 1]

[无须下载，浏览器就能完成接口调试，快来试试][format_png 11 1]

![format_png 12][]

![format_png 13][]

戳这儿

![format_png 14][]

[format_png]: /images/20221024/a4962309c5bd424ea4cba5b59667d6c5.png
[format_png 1]: /images/20221024/1ee6a0c1229e4af4a159719f1568cb06.png
[format_png 2]: /images/20221024/2bf046ddd7684de29850dcf1067575f8.png
[format_png 3]: /images/20221024/0b30aa6d2e1e4bcda1da564f096c8565.png
[format_png 4]: /images/20221024/5afc00d59d85430fac0d2e0ab7e8f9ec.png
[format_png 5]: /images/20221024/39b00d504fe14944a26a306cdc26c77b.png
[format_png 6]: /images/20221024/bd4ce34b12b741caa6c996c837e431d2.png
[format_png 7]: /images/20221024/d2063c2f231f4c8eb46d8cf01089bc0b.png
[format_png 8]: /images/20221024/4da38f1166ee4e738f38b39eb30d2d32.png
[format_png 8 1]: http://mp.weixin.qq.com/s?__biz=MzU2NDc4MjE2Ng%3D%3D&chksm=fc44f721cb337e372ad0582c41717f001c955430f785df3fa7d7a16ab119bfb28a1a1fc7388a&idx=1&mid=2247483928&scene=21&sn=559e77428977c14be5480d374e42de06#wechat_redirect
[format_png 9]: /images/20221024/596720acb1ac48f689e06af8b8596c54.png
[format_png 9 1]: http://mp.weixin.qq.com/s?__biz=MzU2NDc4MjE2Ng%3D%3D&chksm=fc44f1f3cb3378e553ae4b5127fc700c9c090f59565bb2bd294200912fb04936affeaef1eed7&idx=1&mid=2247484618&scene=21&sn=d16198df473eaa704fb75c7531337a31#wechat_redirect
[format_png 10]: /images/20221024/3beacf25b68d4e64b3f4129bc418fa3b.png
[format_png 10 1]: http://mp.weixin.qq.com/s?__biz=MzU2NDc4MjE2Ng%3D%3D&chksm=fc44fe49cb33775f115a52aa902244c0eb00095c50e940f0102b72437cad11a8e4dcc3a5d667&idx=1&mid=2247486320&scene=21&sn=5e7685fa64ac20fef4d9dca35f62e241#wechat_redirect
[format_png 11]: /images/20221024/3f15825c84364cf386b69c7e81e272be.png
[format_png 11 1]: http://mp.weixin.qq.com/s?__biz=MzU2NDc4MjE2Ng%3D%3D&chksm=fc44ffffcb3376e98f2df30f18b30df060a9cf254da0b642b39f018dae0824b88e80a70efa1a&idx=1&mid=2247486150&scene=21&sn=99951710d32e6fd434eb15df35a9283c#wechat_redirect
[format_png 12]: /images/20221024/c59fb3bd8f7d425ba39bde0f01c68eed.png
[format_png 13]: /images/20221024/9ac6af5261e941f08db95d8d7f2a6a46.png
[format_png 14]: /images/20221024/a93a7bdf33cf4d70ac8c745334839254.png