什么鬼,面试官竟然让我用Redis实现一个消息队列!!?
GitHub 9.4k Star 的Java工程師成神之路 ,不來了解一下嗎?
GitHub 9.4k Star 的Java工程師成神之路 ,真的不來了解一下嗎?
GitHub 9.4k Star 的Java工程師成神之路 ,真的確定不來了解一下嗎?
眾所周知,redis是一個高性能的key-value數據庫,在NoSQL數據庫市場上,redis自己就占據了將近半壁江山,足以見到其強大之處。同時,由于redis的單線程特性,我們可以將其用作為一個消息隊列。本篇文章就來講講如何將redis整合到spring boot中,并用作消息隊列的……
一、什么是消息隊列
“消息隊列”是在消息的傳輸過程中保存消息的容器。——《百度百科》
消息我們可以理解為在計算機中或在整個計算機網絡中傳遞的數據。
隊列是我們在學習數據結構的時候學習的基本數據結構之一,它具有先進先出的特性。
所以,消息隊列就是一個保存消息的容器,它具有先進先出的特性。
為什么會出現消息隊列?
下面一張圖我們來簡單了解一下消息隊列
由上圖可以看到,消息隊列充當了一個中間人的角色,我們可以通過操作這個消息隊列來保證我們的系統穩定。
二、環境準備
Java環境:jdk1.8
spring boot版本:2.2.1.RELEASE
redis-server版本:3.2.100
三、相關依賴
這里只展示與redis相關的依賴,
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-redis</artifactId> </dependency> <dependency><groupId>org.springframework.integration</groupId><artifactId>spring-integration-redis</artifactId> </dependency>這里解釋一下這兩個依賴:
- 第一個依賴是對redis NoSQL的支持
- 第二個依賴是spring integration與redis的結合,這里添加這個代碼主要是為了實現分布式鎖
四、配置文件
這里只展示與redis相關的配置
redis所在的的地址 spring.redis.host=localhost redis數據庫索引,從0開始,可以從redis的可視化客戶端查看 spring.redis.database=1 redis的端口,默認為6379 spring.redis.port=6379 redis的密碼 spring.redis.password= 連接redis的超時時間(ms),默認是2000 spring.redis.timeout=5000 連接池最大連接數 spring.redis.jedis.pool.max-active=16 連接池最小空閑連接 spring.redis.jedis.pool.min-idle=0 連接池最大空閑連接 spring.redis.jedis.pool.max-idle=16 連接池最大阻塞等待時間(負數表示沒有限制) spring.redis.jedis.pool.max-wait=-1 連接redis的客戶端名 spring.redis.client-name=mall五、代碼配置
redis用作消息隊列,其在spring boot中的主要表現為一個RedisTemplate.convertAndSend()方法和一個MessageListener接口。所以我們要在IOC容器中注入一個RedisTemplate和一個實現了MessageListener接口的類。話不多說,先看代碼
配置RedisTemplate
配置RedisTemplate的主要目的是配置序列化方式以解決亂碼問題,同時合理配置序列化方式還能降低一點性能開銷。
/*** 配置RedisTemplate,解決亂碼問題*/ @Bean public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) {LOGGER.debug("redis序列化配置開始");RedisTemplate<String, Object> template = new RedisTemplate<>();template.setConnectionFactory(factory);// string序列化方式RedisSerializer serializer = new GenericJackson2JsonRedisSerializer();// 設置默認序列化方式template.setDefaultSerializer(serializer);template.setKeySerializer(new StringRedisSerializer());template.setHashValueSerializer(serializer);LOGGER.debug("redis序列化配置結束");return template; }代碼第12行,我們配置默認的序列化方式為GenericJackson2JsonRedisSerializer
代碼第13行,我們配置鍵的序列化方式為StringRedisSerializer
代碼第14行,我們配置哈希表的值的序列化方式為GenericJackson2JsonRedisSerializer
RedisTemplate幾種序列化方式的簡要介紹
| StringRedisSerializer | 將對象序列化為字符串,但是經測試,無法序列化對象,一般用在key上 |
| OxmSerializer | 將對象序列化為xml性質,本質上是字符串 |
| ByteArrayRedisSerializer | 默認序列化方式,將對象序列化為二進制字節,但是需要對象實現Serializable接口 |
| GenericFastJsonRedisSerializer | json序列化,使用fastjson序列化方式序列化對象 |
| GenericJackson2JsonRedisSerializer | json序列化,使用jackson序列化方式序列化對象 |
六、redis隊列監聽器(消費者)
上面說了,與redis隊列監聽器相關的類為一個名為MessageListener的接口,下面是該接口的源碼
public interface MessageListener {void onMessage(Message message, @Nullable byte[] pattern); }可以看到,該接口僅有一個onMessage(Message message, @Nullable byte[] pattern)方法,該方法便是監聽到隊列中消息后的回調方法。下面解釋一下這兩個參數:
- message:redis消息類,該類中僅有兩個方法
- byte[] getBody()以二進制形式獲取消息體
- byte[] getChannel()以二進制形式獲取消息通道
- pattern:二進制形式的消息通道,和message.getChannel()返回值相同
介紹完接口,我們來實現一個簡單的redis隊列監聽器
@Component public class RedisListener implement MessageListener{private static final Logger LOGGER = LoggerFactory.getLogger(RedisListener.class);@Overridepublic void onMessage(Message message,byte[] pattern){LOGGER.debug("從消息通道={}監聽到消息",new String(pattern));LOGGER.debug("從消息通道={}監聽到消息",new String(message.getChannel()));LOGGER.debug("元消息={}",new String(message.getBody()));// 新建一個用于反序列化的對象,注意這里的對象要和前面配置的一樣// 因為我前面設置的默認序列化方式為GenericJackson2JsonRedisSerializer// 所以這里的實現方式為GenericJackson2JsonRedisSerializerRedisSerializer serializer=new GenericJackson2JsonRedisSerializer();LOGGER.debug("反序列化后的消息={}",serializer.deserialize(message.getBody()));} }代碼很簡單,就是輸出參數中包含的關鍵信息。需要注意的是,RedisSerializer的實現要與上面配置的序列化方式一致。
隊列監聽器實現完以后,我們還需要將這個監聽器添加到redis隊列監聽器容器中,代碼如下:
@Bean public public RedisMessageListenerContainer container(RedisConnectionFactory factory) {RedisMessageListenerContainer container = new RedisMessageListenerContainer();container.setConnectionFactory(factory);container.addMessageListener(redisListener, new PatternTopic("demo-channel"));return container; }這幾行代碼大概意思就是新建一個Redis消息監聽器容器,然后將監聽器和管道名想綁定,最后返回這個容器。
這里要注意的是,這個管道名和下面將要說的推送消息時的管道名要一致,不然監聽器監聽不到消息。
七、redis隊列推送服務(生產者)
上面我們配置了RedisTemplate將要在這里使用到。
代碼如下:
@Service public class Publisher{@Autowriteprivate RedisTemplate redis;public void publish(Object msg){redis.convertAndSend("demo-channel",msg);} }關鍵代碼為第7行,redis.convertAndSend()這個方法的作用為,向某個通道(參數1)推送一條消息(第二個參數)。
這里還是要注意上面所說的,生產者和消費者的通道名要相同。
至此,消息隊列的生產者和消費者已經全部編寫完成。
八、遇到的問題及解決辦法
1、spring boot使用log4j2日志框架問題
在我添加了spring-boot-starter-log4j2依賴并在spring-boot-starter-web中排除了spring-boot-starter-logging后,運行項目,還是會提示下面的錯誤:
SLF4J: Class path contains multiple SLF4J bindings. SLF4J: Found binding in [jar:file:.....m2/repository/ch/qos/logback/logback-classic/1.2.3/logback-classic-1.2.3.jar!/org/slf4j/impl/StaticLoggerBinder.class] SLF4J: Found binding in [jar:file:.....m2/repository/org/apache/logging/log4j/log4j-slf4j-impl/2.12.1/log4j-slf4j-impl-2.12.1.jar!/org/slf4j/impl/StaticLoggerBinder.class] SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation. SLF4J: Actual binding is of type [ch.qos.logback.classic.util.ContextSelectorStaticBinder]這個錯誤就是maven中有多個日志框架導致的。后來通過依賴分析,發現在spring-boot-starter-data-redis中,也依賴了spring-boot-starter-logging,解決辦法也很簡單,下面貼出詳細代碼
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-redis</artifactId><exclusions><exclusion><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-logging</artifactId></exclusion></exclusions> </dependency> <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId><exclusions><exclusion><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-logging</artifactId></exclusion></exclusions> </dependency> <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-log4j2</artifactId> </dependency> <dependency><groupId>org.springframework.integration</groupId><artifactId>spring-integration-redis</artifactId> </dependency>2、redis隊列監聽器線程安全問題
redis隊列監聽器的監聽機制是:使用一個線程監聽隊列,隊列有未消費的消息則取出消息并生成一個新的線程來消費消息。如果你還記得,我開頭說的是由于redis單線程特性,因此我們用它來做消息隊列,但是如果監聽器每次接受一個消息就生成新的線程來消費信息的話,這樣就完全沒有使用到redis的單線程特性,同時還會產生線程安全問題。
單一消費者(一個通道只有一個消費者)的解決辦法
最簡單的辦法莫過于為onMessage()方法加鎖,這樣簡單粗暴卻很有用,不過這種方式無法控制隊列監聽的速率,且無限制的創造線程最終會導致系統資源被占光。
那如何解決這種情況呢?線程池。
在將監聽器添加到容器的配置的時候,RedisMessageListenerContainer類中有一個方法setTaskExecutor(Executor taskExecutor)可以為監聽容器配置線程池。配置線程池以后,所有的線程都會由該線程池產生,由此,我們可以通過調節線程池來控制隊列監聽的速率。
多個消費者(一個通道有多個消費者)的解決辦法
單一消費者的問題相比于多個消費者來說還是較為簡單,因為Java內置的鎖都是只能控制自己程序的運行,不能干擾其他的程序的運行;然而現在很多時候我們都是在分布式環境下進行開發,這時處理多個消費者的情況就很有意義了。
那么這種問題如何解決呢?分布式鎖。
下面來簡要科普一下什么是分布式鎖:
分布式鎖是指在分布式環境下,同一時間只有一個客戶端能夠從某個共享環境中(例如redis)獲取到鎖,只有獲取到鎖的客戶端才能執行程序。
然后分布式鎖一般要滿足:排他性(即同一時間只有一個客戶端能夠獲取到鎖)、避免死鎖(即超時后自動釋放)、高可用(即獲取或釋放鎖的機制必須高可用且性能佳)
上面講依賴的時候,我們導入了一個spring-integration-redis依賴,這個依賴里面包含了很多實用的工具類,而我們接下來要講的分布式鎖就是這個依賴下面的一個工具包RedisLockRegistry。
首先講一下如何使用,導入了依賴以后,首先配置一個Bean
@Bean public RedisLockRegistry redisLockRegistry(RedisConnectionFactory factory) {return new RedisLockRegistry(factory, "demo-lock",60); }RedisLockRegistry的構造函數,第一個參數是redis連接池,第二個參數是鎖的前綴,即取出的鎖,鍵名為“demo-lock:KEY_NAME”,第三個參數為鎖的過期時間(秒),默認為60秒,當持有鎖超過該時間后自動過期。
使用鎖的方法,下面是對監聽器的修改
@Component public class RedisListener implement MessageListener{@Autowriteprivate RedisLockRegistry redisLockRegistry;private static final Logger LOGGER = LoggerFactory.getLogger(RedisListener.class);@Overridepublic void onMessage(Message message,byte[] pattern){Lock lock=redisLockRegistry.obtain("lock");try{lock.lock(); //上鎖LOGGER.debug("從消息通道={}監聽到消息",new String(pattern));LOGGER.debug("從消息通道={}監聽到消息",new String(message.getChannel()));LOGGER.debug("元消息={}",new String(message.getBody()));// 新建一個用于反序列化的對象,注意這里的對象要和前面配置的一樣// 因為我前面設置的默認序列化方式為GenericJackson2JsonRedisSerializer// 所以這里的實現方式為GenericJackson2JsonRedisSerializerRedisSerializer serializer=new GenericJackson2JsonRedisSerializer();LOGGER.debug("反序列化后的消息={}",serializer.deserialize(message.getBody()));} catch (Exception e) {e.printStackTrace();} finally {lock.unlock(); //解鎖}} }上面代碼的代碼比起前面的監聽器代碼,只是多了一個注入的RedisLockRegistry,一個通過redisLockRegistry.obtain()方法獲取鎖,一個加鎖一個解鎖,然后這就完成了分布式鎖的使用。
注意這個獲取鎖的方法redisLockRegistry.obtain(),其返回的是一個名為RedisLock的鎖,這是一個私有內部類,它實現了Lock接口,因此我們不能從代碼外部創建一個他的實例,只能通過obtian()方法來獲取這個鎖。
新人創作打卡挑戰賽發博客就能抽獎!定制產品紅包拿不停!總結
以上是生活随笔為你收集整理的什么鬼,面试官竟然让我用Redis实现一个消息队列!!?的全部內容,希望文章能夠幫你解決所遇到的問題。
- 上一篇: NYOJ 622 Vote
- 下一篇: NYOJ 623 A*B Problem