快捷導(dǎo)航

SpringBoot集成Redis實現(xiàn)消息隊列的方法

更新時間：2021年02月10日 09:31:20 作者：新猿一馬

這篇文章主要介紹了SpringBoot集成Redis實現(xiàn)消息隊列的方法，文中通過示例代碼介紹的非常詳細，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧

list 原理說明

Redis 的 list 是按照插入順序排序的字符串鏈表。

如圖所示，可以通過 lpush 和 rpop 或者 rpush 和 lpop 實現(xiàn)消息隊列。

1 lpush 和 rpop

2 rpush 和 lpop

消息隊列功能實現(xiàn)

引入 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發(fā)送和接收接口

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

MQ發(fā)送和接收實現(xiàn)類

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發(fā)送和接收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();
    }
  }
 
}

消息隊列功能測試

調(diào)用 http://localhost:8080/api/produce 接口往隊列里面添加 a、b、c、d元素。

調(diào)用 http://localhost:8080/api/consume 消費隊列里面的元素。

從截圖我們可以看到，即使當(dāng)隊列為空，消費者依然在不停的 pop 數(shù)據(jù)，這就是浪費生命的空輪詢。

那如何解決這個空輪詢的問題呢？

你也許會想使用 Thread.sleep() 讓消費者線程隔一段時間再消費。

使用 Thread.sleep() 會有什么問題么？

A 如果生產(chǎn)者速度大于消費者消費速度，消息隊列長度會一直增大，時間久了會占用大量內(nèi)存空間。

B 如果睡眠時間過長，這樣不能處理一些時效性的消息，睡眠時間過短，也會在連接上造成比較大的開銷。

有沒有更優(yōu)雅和更合適的方式呢？

brpop 和 blpop 實現(xiàn)阻塞讀取，下面以 blpop 為例來說明問題。

blpop 理論說明

blpop 命令

blpop key1...keyN timeout

blpop 說明

blpop 是阻塞式列表的彈出原語。當(dāng)給定列表內(nèi)沒有任何元素可供彈出的時候，連接將被 blpop 命令阻塞。直到有另一個客戶端對給定的這些 key 的任意一個執(zhí)行 lpush 或 rpush 命令為止。

當(dāng)給定多個 key 參數(shù)時，按參數(shù) key 的先后順序依次檢查各個列表，彈出第一個非空列表的頭元素。

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

timeout：阻塞隊列超時時間。

blpop 代碼實現(xiàn)

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