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

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

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; @Configurationpublic 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ā)送和接收實(shí)現(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; @Servicepublic 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(); } } }消息隊(duì)列功能測(cè)試

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

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

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

那如何解決這個(gè)空輪詢的問(wèn)題呢？

你也許會(huì)想使用 Thread.sleep() 讓消費(fèi)者線程隔一段時(shí)間再消費(fèi)。

使用 Thread.sleep() 會(huì)有什么問(wèn)題么？

A 如果生產(chǎn)者速度大于消費(fèi)者消費(fèi)速度，消息隊(duì)列長(zhǎng)度會(huì)一直增大，時(shí)間久了會(huì)占用大量?jī)?nèi)存空間。

B 如果睡眠時(shí)間過(guò)長(zhǎng)，這樣不能處理一些時(shí)效性的消息，睡眠時(shí)間過(guò)短，也會(huì)在連接上造成比較大的開銷。

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

brpop 和 blpop 實(shí)現(xiàn)阻塞讀取，下面以 blpop 為例來(lái)說(shuō)明問(wèn)題。

blpop 命令

blpop key1...keyN timeout

blpop 說(shuō)明

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

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

key1...keyN：表示不同的隊(duì)列名。

timeout：阻塞隊(duì)列超時(shí)時(shí)間。

blpop 代碼實(shí)現(xiàn)

public void blockingConsume() { List<Object> obj = redisTemplate.executePipelined(new RedisCallback<Object>() { @Nullable @Override public Object doInRedis(RedisConnection connection) throws DataAccessException { //隊(duì)列沒(méi)有元素會(huì)阻塞操作，直到隊(duì)列獲取新的元素或超時(shí) return connection.bLPop(TIME_OUT, MESSAGE_KEY.getBytes()); } },new StringRedisSerializer()); for (Object str: obj) { log.info('blockingConsume : {}', str); }}

阻塞線程每隔10s超時(shí)執(zhí)行一次。該方法解決了 CPU 空轉(zhuǎn)的問(wèn)題。

到此這篇關(guān)于SpringBoot集成Redis實(shí)現(xiàn)消息隊(duì)列的方法的文章就介紹到這了,更多相關(guān)SpringBoot Redis消息隊(duì)列內(nèi)容請(qǐng)搜索好吧啦網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持好吧啦網(wǎng)！