Springboot+Redis(发布订阅模式)跨多服务器实战
2024-09-05 22:15:48
一:redis中发布订阅功能(http://www.redis.cn/commands.html#pubsub)
PSUBSCRIBE pattern [pattern …]:订阅一个或者多个符合pattern格式的频道
PUBLISH channel message:发布消息到chanel中
PUBSUB subcommand [argument [argument …]]:查看订阅与发布系统状态
PUNSUBSCRIBE [pattern [pattern …]]:退订所有符合格式的频道
SUBSCRIBE channel [channel …]:订阅一个或者多个频道
UNSUBSCRIBE [channel [channel …]]:取消订阅频道
二:实战使用reids中的发布订阅模式解决部署在阿里服务与本地后台服务的接口调用不通问题(跨多服务器)
当然也可使用更为强大的消息中间件(RabbitMQ、ActiveMQ、RocketMQ、Kafka、ZeroMQ)
1:redis使用到的maven依赖
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
<!-- springboot1.5版本使用jedis,2.0以上版本使用lettuce,本项目使用jedis,所以需要排除lettuce -->
<exclusions>
<exclusion>
<groupId>redis.clients</groupId>
<artifactId>jedis</artifactId>
</exclusion>
<!--
<exclusion>
<groupId>io.lettuce</groupId>
<artifactId>lettuce-core</artifactId>
</exclusion>
-->
</exclusions>
</dependency>
2:application.yml配置
redis:
host: localhost
port: 8379
password: 123456
database: 2
timeout: 50s
# 如果使用的jedis 则将lettuce改成jedis即可
jedis:
pool:
max-active: 8
max-idle: 8
min-idle: 0
3:publisher发布者发布消息
/**
* 建立发布者,通过频道发布消息
* @param key 发布者
* @param value 消息
*/
public void publish(String key,Object value){
this.redisTemplate.convertAndSend(key,value);
}
redisUtils.publish(RedisTopicEnums.TOPIC_DISCOVER.getTopic(),message);
4:第一种实现方法
/**
* redis消息监听器容器
* @param connectionFactory
* @param healthyListenerAdapter 健康扫描消息订阅处理器
* @param settingsListenerAdapter 配置健康扫描消息订阅处理器
* @return
*/
@Bean
RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,
MessageListenerAdapter healthyListenerAdapter,
MessageListenerAdapter settingsListenerAdapter
) {
RedisMessageListenerContainer container = new RedisMessageListenerContainer();
container.setConnectionFactory(connectionFactory);
//设备健康扫描绑定消息订阅处理器
container.addMessageListener(healthyListenerAdapter, new PatternTopic("healthy_topic"));
//设备配置扫描并绑定消息订阅处理器
container.addMessageListener(settingsListenerAdapter, new PatternTopic("settings_topic"));
return container;
} /**
* 设备健康消息订阅处理器,并指定处理方法(利用反射的机制调用消息处理器的业务方法)
* @param receiver
* @return
*/
@Bean
MessageListenerAdapter healthyListenerAdapter(ReceiverRedisMessage receiver) {
MessageListenerAdapter messageListenerAdapter = new MessageListenerAdapter(receiver, "healthy");
return messageListenerAdapter;
} /**
* 设备健康消息订阅处理器,并指定处理方法
* @param receiver
* @return
*/
@Bean
MessageListenerAdapter settingsListenerAdapter(ReceiverRedisMessage receiver) {
MessageListenerAdapter messageListenerAdapter = new MessageListenerAdapter(receiver, "settings");
return messageListenerAdapter;
}
缺点:1:考虑实际运用中可能会订阅多个主题,每加一个主题(Topic)都需要使用container.addMessageListener(listenerAdapter,
new
PatternTopic(
"topic"
));不合理
2:考虑到后期尽量不该变原有代码进行扩展,推荐使用下面第二种方式实现(保证开闭原则)
5:第二种实现方法:定义订阅者接收消息器接口
/**
* 订阅者接收消息的基类
* @author : ywb
* @createdDate : 2020/8/6
* @updatedDate
*/
public interface Subscriber extends MessageListener { /**
* 类型
* @return
*/
default String getType() {
return this.getClass().getSimpleName();
} /**
* 通道名称
* @return
*/
String getTopic(); }
6:定义不同主题枚举类型,后期增加一个管道,增加一个枚举信息即可
/**
* 定义不同主题类型
* @author : ywb
* @createdDate : 2020/8/7
* @updatedDate
*/
public enum RedisTopicEnums { /**
* redis主题名称定义 需要与发布者一致
*
*/
TOPIC_DISCOVERY("topic:discovery", "设备发现变更Topic"), TOPIC_HEALTHY("topic:healthy", "健康扫描的设备Topic"), TOPIC_SETTINGS("topic:settings", "配置扫描变更的设备Topic"), TOPIC_DISCOVER("topic:discover", "发现设备Topic"), ;
/**
* 主题名称
*/
private String topic; /**
* 描述
*/
private String description; RedisTopicEnums(String topic, String description) {
this.topic = topic;
this.description = description;
} public String getTopic() {
return topic;
} public String getDescription() {
return description;
} }
7:实现多个订阅者,后续增加一个订阅者,只需要多加上一个订阅者类,从而不用改动redis消息 监听容器配置
7.1:设备健康扫描订阅者
/**
* 设备健康扫描的订阅者
*
* @author : ywb
* @createdDate : 2020/8/7
* @updatedDate
*/
@Component
@Slf4j
public class HealthySubscriber implements Subscriber { @Autowired
private DeviceService deviceService; @Autowired
private RedisTemplate<String, Object> redisTemplate; @Override
public String getTopic() {
return RedisTopicEnums.TOPIC_HEALTHY.getTopic();
} @Override
public void onMessage(Message message, byte[] pattern) { String deviceIds = (String) redisTemplate.getValueSerializer().deserialize(message.getBody()); log.info(">> 订阅消息,设备健康异常编号:{}", deviceIds); // TODO 这里是收到通道的消息之后执行的方法
String[] split = deviceIds.split(","); Map<String, Set<Integer>> idsMap = TokenSplit.getDeviceIdRegex(split); for (Map.Entry<String, Set<Integer>> stringSetEntry : idsMap.entrySet()) {
DeviceHandle healthyHandle = new DeviceHealthyHandle();
healthyHandle.respondHandle(stringSetEntry.getValue());
} }
}
7.2:配置扫描订阅者
/**
* 设备配置变更订阅者
*
* @author : ywb
* @createdDate : 2020/8/7
* @updatedDate
*/
@Component
@Slf4j
public class SettingsSubscriber implements Subscriber { @Autowired
private RedisTemplate<String, Object> redisTemplate; @Override
public String getTopic() {
return RedisTopicEnums.TOPIC_SETTINGS.getTopic();
} @Override
public void onMessage(Message message, byte[] pattern) { //使用redis convertAndSend发布消息,订阅者获取字符串字节必须要反序列
String deviceIds = (String) redisTemplate.getValueSerializer().deserialize(message.getBody()); log.info(">>订阅消息,设备配置变更编号:{}", deviceIds); // TODO 这里是收到通道的消息之后执行的方法
String[] split = deviceIds.split(","); Map<String, Set<Integer>> idsMap = TokenSplit.getDeviceIdRegex(split); for (Map.Entry<String, Set<Integer>> stringSetEntry : idsMap.entrySet()) {
DeviceScannerHandle scannerHandle = new DeviceScannerHandle();
scannerHandle.respondHandle(stringSetEntry.getValue());
}
}
}
8:redisConfig配置,消息监听器容器配置
@Configuration
public class RedisConfig { /**
* 自定义 redisTemplate<String, Object>
*
* @param redisConnectionFactory
* @return
*/
@Bean
public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) {
RedisTemplate<String, Object> template = new RedisTemplate<>();
template.setConnectionFactory(redisConnectionFactory); ObjectMapper om = new ObjectMapper();
// 指定要序列化的域,field,get和set,以及修饰符范围,ANY是都有包括private和public
om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
// 指定序列化输入的类型,类必须是非final修饰的,final修饰的类
// om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
// om.activateDefaultTyping(BasicPolymorphicTypeValidator.builder().build(), ObjectMapper.DefaultTyping.EVERYTHING);
om.activateDefaultTyping(new LaissezFaireSubTypeValidator(), ObjectMapper.DefaultTyping.NON_FINAL);
om.setDateFormat(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS"));
om.setTimeZone(TimeZone.getTimeZone("GMT+8"));
// 不转换值为 null 的对象
// om.setSerializationInclusion(JsonInclude.Include.NON_NULL); Jackson2JsonRedisSerializer<Object> jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer<>(Object.class);
jackson2JsonRedisSerializer.setObjectMapper(om); // key 采用 string 的序列化方式
template.setKeySerializer(new StringRedisSerializer());
// value 采用 jackson 的序列化方式
template.setValueSerializer(jackson2JsonRedisSerializer);
// hash 的 key 采用 string 的序列化方式
template.setHashKeySerializer(new StringRedisSerializer());
// hash 的 value 采用 jackson 的序列化方式
template.setHashValueSerializer(jackson2JsonRedisSerializer);
template.afterPropertiesSet(); return template;
} /**
* DependencyDescriptor
* 重点
* 首先判断注入的类型,如果是数组、Collection、Map,则注入的是元素数据,即查找与元素类型相同的Bean,注入到集合中。
* 强调下Map类型,Map的 key 为Bean的 name,value 为 与定义的元素类型相同的Bean。
*将所有相同类型(实现了同一个接口)的Bean,一次性注入到集合类型中,具体实现查看spring源码
*
* 获取Subscriptor接口所有的实现类
* 注入所有实现了接口的Bean
* 将所有的配置消息接收处理类注入进来,那么消息接收处理类里面的注解对象也会注入进来
*/
@Autowired
private transient List<Subscriber> subscriptorList; @Bean
RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory) { //创建一个消息监听对象
RedisMessageListenerContainer container = new RedisMessageListenerContainer(); //将监听对象放入到容器中
container.setConnectionFactory(connectionFactory); if (this.subscriptorList != null && this.subscriptorList.size() > 0) {
for (Subscriber subscriber : this.subscriptorList) { if (subscriber == null || StringUtils.isBlank(subscriber.getTopic())) {
continue;
}
//一个订阅者对应一个主题通道信息
container.addMessageListener(subscriber, new PatternTopic(subscriber.getTopic()));
}
} return container;
}
注:编写代码要善于运用设计模式
最新文章
- Ambari server:无法显示内存,CPU等使用率
- C语言 复杂队列(链表队列)
- 今天上传公司服务器出现的.net framework版本错误问题
- javascript技术难点之this、new、apply和call详解
- SVN基于Maven的Web项目更新,本地过程详细解释
- C面试题
- C#异步编程(async and await)及异步方法同步调用
- Android开发之漫漫长途 XIV——RecyclerView
- Atcoder E - Knapsack 2 (01背包进阶版 ex )
- (转)nginx uwsgi wsgi django 这些东西究竟是什么关系
- ssh整合not found class 异常总结
- String的intern()方法详解
- LeetCode题解之 Increasing Order Search Tree
- ATmega8仿真——键盘扫描的学习
- uva 11525(线段树)
- php array key 的存储规则
- 使用通配符配置action
- Java Socket实现基于TCP和UDP多线程通信
- docker安装总结 linux红帽系列
- linux crontab 常用时间设置
热门文章
- UEFI、BIOS、GPT、MBR等概念的辨析
- docker 停止、启动、删除镜像指令
- jzoj1497. 景点中心
- 《SeleniumBasic 3.141.0.0 - 在VBA中操作浏览器》系列文章之一:SeleniumBasic的下载
- Java简介以及入门
- NGINX 负载均衡的理解
- [程序员代码面试指南]递归和动态规划-数字字符串转换为字母组合的种数(DP)
- RabbitMQ消息积压的几种解决思路
- Centos6.5 离线 Openssh 升级
- robotframework获取Token