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); } }
阻塞線程每隔10s超時執(zhí)行一次。該方法解決了 CPU 空轉(zhuǎn)的問題。
到此這篇關(guān)于SpringBoot集成Redis實現(xiàn)消息隊列的方法的文章就介紹到這了,更多相關(guān)SpringBoot Redis消息隊列內(nèi)容請搜索服務(wù)器之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持服務(wù)器之家！
原文鏈接：https://blog.csdn.net/jack1liu/article/details/113725818
