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 空轉的問題，