Spring+Redis 進行釋出訂閱
阿新 • • 發佈:2018-12-13
訊息是許多軟體架構的重要組成部分。訊息傳遞解決方案提供了高效能,可擴充套件性,佇列永續性和耐用性,故障轉移支援等,以及許多更漂亮具備的功能,在Java世界中大多總是使用JMS實現。後來使用Apache ActiveMQ的,有時只是需要簡單的排隊支援,而Apache ActiveMQ顯得過於複雜。事實上,Redis 不僅提供一個NoSQL資料庫,同時提供了一套訊息系統。
使用Redis作為pub/sub的佇列,將訊息存放在Redis記憶體中。
話不多說直接上配置
yml配置
編寫監聽類來接收訂閱的訊息
Cat監聽
package com.maoxs.listener;
import org. springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
/**
* 監聽傳送的訊息
*/
@Component
public class CatListener implements MessageListener {
@Autowired
RedisTemplate redisTemplate;
@Override
public void onMessage(Message message, byte[] bytes) {
System.out.println("我是Cat監聽" + message.toString());
}
}
Fish監聽
package com.maoxs.listener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.core.RedisTemplate;
/**
* 監聽傳送的訊息
*/
public class FishListener implements MessageListener {
@Autowired
RedisTemplate redisTemplate;
@Override
public void onMessage(Message message, byte[] bytes) {
System.out.println("我是Fish監聽" + message.toString());
}
}
Redis的相關配置
package com.maoxs.redis;
import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.maoxs.listener.CatListener;
import com.maoxs.listener.FishListener;
import com.maoxs.pojo.MessageReceiver;
import org.springframework.cache.CacheManager;
import org.springframework.cache.annotation.CachingConfigurerSupport;
import org.springframework.cache.annotation.EnableCaching;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.cache.RedisCacheManager;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import java.util.Arrays;
@SuppressWarnings({"all"})
@Configuration
@EnableCaching
public class CacheConfig extends CachingConfigurerSupport {
/**
* redis訊息監聽器容器
* 可以新增多個監聽不同話題的redis監聽器,只需要把訊息監聽器和相應的訊息訂閱處理器繫結,該訊息監聽器
* 通過反射技術呼叫訊息訂閱處理器的相關方法進行一些業務處理
* @param connectionFactory
* @param listenerAdapter
* @return
*/
@Bean
//相當於xml中的bean
RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,
MessageListenerAdapter CatAdapter, MessageListenerAdapter FishAdapter) {
RedisMessageListenerContainer container = new RedisMessageListenerContainer();
container.setConnectionFactory(connectionFactory);
//訂閱了一個叫chat 的通道
container.addMessageListener(CatAdapter, new PatternTopic("cat"));
container.addMessageListener(FishAdapter, new PatternTopic("fish"));
//這個container 可以新增多個 messageListener
return container;
}
/**
* 訊息監聽器介面卡,繫結訊息處理器
*
* @param receiver
* @return
*/
@Bean
MessageListenerAdapter CatAdapter() {
return new MessageListenerAdapter(new CatListener());
}
/**
* 訊息監聽器介面卡,繫結訊息處理器
*
* @param receiver
* @return
*/
@Bean
MessageListenerAdapter FishAdapter() {
return new MessageListenerAdapter(new FishListener());
}
/**
* redis 讀取內容的template
*/
@Bean
StringRedisTemplate stringRedisTemplate(RedisConnectionFactory connectionFactory) {
return new StringRedisTemplate(connectionFactory);
}
@Bean
public RedisTemplate<String, String> redisTemplate(RedisConnectionFactory factory) {
StringRedisTemplate template = new StringRedisTemplate(factory);
//定義value的序列化方式
Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
ObjectMapper om = new ObjectMapper();
om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
jackson2JsonRedisSerializer.setObjectMapper(om);
template.setValueSerializer(jackson2JsonRedisSerializer);
template.setHashKeySerializer(jackson2JsonRedisSerializer);
template.afterPropertiesSet();
return template;
}
}
建立一個定時器模擬釋出訊息
package com.maoxs.service;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
@EnableScheduling //開啟定時器功能
@Component
public class MessageSender {
@Autowired
@Resource(name = "stringRedisTemplate")
private StringRedisTemplate stringRedisTemplate;
@Scheduled(fixedRate = 2000) //間隔2s 通過StringRedisTemplate物件向redis訊息佇列cat頻道釋出訊息
public void sendCatMessage(){
stringRedisTemplate.convertAndSend("cat","貓");
}
@Scheduled(fixedRate = 1000) //間隔1s 通過StringRedisTemplate物件向redis訊息佇列fish頻道釋出訊息
public void sendFishMessage(){
stringRedisTemplate.convertAndSend("fish","魚");
}
}
最後就是啟動類
package com.maoxs;
import org.mybatis.spring.annotation.MapperScan;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.ComponentScan;
@SpringBootApplication
@ComponentScan("com.maoxs")
public class DemoApplication {
private static final Logger LOGGER = LoggerFactory.getLogger(DemoApplication.class);
public static void main(String[] args) {
SpringApplication.run(DemoApplication.class, args);
}
}
然後就啟動看下效果
ok 大功告成!!
相關連線:
注:如果不對聯絡本寶寶及時改正~~