springboot集成redis实现消息发布订阅模式-双通道(跨多服务器)

环境：SpringBoot + jdk1.8

基础配置参考
https://blog.csdn.net/llll234/article/details/80966952

查看了基础配置那么会遇到一下几个问题：

1.实际应用中可能会订阅多个通道，而一下这种写法不太通用
container.addMessageListener(listenerAdapter(new RedisPmpSub()),new PatternTopic("pmp"));

2.使用过程中使用new RedisPmpSub()配置消息接收对象会有问题。
如果RedisPmpSub既是消息接收类，也是消息处理类。那么如果此时需要注入Bean，会成功吗？

3.考虑后期的扩展性是否能尽量不改变原有代码的基础上，进行扩展

额外的配置文件

<dependency>

    <groupId>org.projectlombok</groupId>

    <artifactId>lombok</artifactId>

    <optional>true</optional>

</dependency>

<dependency>

    <groupId>commons-lang</groupId>

    <artifactId>commons-lang</artifactId>

    <version>RELEASE</version>

</dependency>

<dependency>

    <groupId>com.google.code.gson</groupId>

    <artifactId>gson</artifactId>

</dependency>

由于GsonUtil依赖的是某个SDK，GsonUtil.toJson(this, BasePubMessage.class)可替换为
new Gson().toJson(this, BasePubMessage.class);
lombok需要下载插件

发布者

枚举定义

考虑到可维护性，采用枚举的方式定义管道RedisChannelEnums

 public enum RedisChannelEnums {

     /**redis频道code定义 需要与发布者一致*/

     LIVE_INFO_CHANGE("LIVE_INFO_CHANGE","直播信息改变"),

     ;

     /** 枚举定义+描述 */

     private String code;

     private String description;

     RedisChannelEnums(String code, String description) {

         this.code = code;

         this.description = description;

     }

     /** 根据code获取对应的枚举对象 */

     public static RedisChannelEnums getEnum(String code) {

         RedisChannelEnums[] values = RedisChannelEnums.values();

         if (null != code && values.length > 0) {

             for (RedisChannelEnums value : values) {

                 if (value.code == code) {

                     return value;

                 }

             }

         }

         return null;

     }

     /** 该code在枚举列表code属性是否存在 */

     public static boolean containsCode(String code) {

         RedisChannelEnums anEnum = getEnum(code);

         return anEnum != null;

     }

     /** 判断code与枚举中的code是否相同 */

     public static boolean equals(String code, RedisChannelEnums calendarSourceEnum) {

         return calendarSourceEnum.code == code;

     }

     public String getCode() {

         return code;

     }

     public String getDescription() {

         return description;

     }

 }

消息模板

为了兼容不同的业务场景，需要定义消息模板对象BasePubMessage
其中ToString方法的作用是将对象转成Json字符

 @Data

 public abstract class BasePubMessage {

     /**发布订阅频道名称*/

     protected String channel;

     protected String extra;

     @Override

     public String toString() {

         return GsonUtil.toJson(this, BasePubMessage.class);

     }

 }

消息对象LiveChangeMessage
其中ToString方法的作用是将对象转成Json字符

 @Data

 public class LiveChangeMessage extends BasePubMessage {

     /**直播Ids*/

     private String liveIds;

     @Override

     public String toString() {

         return GsonUtil.toJson(this, LiveChangeMessage.class);

     }

 }

发布者服务

public interface RedisPub {

    /**

     * 集成redis实现消息发布订阅模式-双通道

     * @param redisChannelEnums 枚举定义

     * @param basePubMessage 消息

     */

    void sendMessage(RedisChannelEnums redisChannelEnums, BasePubMessage basePubMessage);

}

 @Service

 public class RedisPubImpl implements RedisPub {

     @Resource

     private StringRedisTemplate stringRedisTemplate;

     @Override

     public void sendMessage(RedisChannelEnums redisChannelEnums, BasePubMessage basePubMessage) {

         if(redisChannelEnums ==null || basePubMessage ==null){

             return;

         }

         basePubMessage.setChannel(redisChannelEnums.getCode());

         stringRedisTemplate.convertAndSend(redisChannelEnums.getCode(), basePubMessage.toString());

         System.out.println("发布成功！");

     }

 }

订阅者

注解配置

RedisConfig作为订阅者的配置类，主要作用是：Redis消息监听器容器、配置消息接收处理类
同时新加入的功能解决了我们上面提出的几个问题

 @Service

 @Configuration

 @EnableCaching

 public class RedisConfig {

     /**

      * 存放策略实例

      * classInstanceMap : key-beanName value-对应的策略实现

      */

     private ConcurrentHashMap<String, BaseSub> classInstanceMap = new ConcurrentHashMap<>(20);

     /**

      * 注入所有实现了Strategy接口的Bean

      *

      * @param strategyMap

      *         策略集合

      */

     @Autowired

     public RedisConfig(Map<String, BaseSub> strategyMap) {

         this.classInstanceMap.clear();

         strategyMap.forEach((k, v) ->

                 this.classInstanceMap.put(k.toLowerCase(), v)

         );

     }

     /**

      * Redis消息监听器容器

      *

      * @param connectionFactory

      *

      * @return

      */

     @Bean

     RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory) {

         RedisMessageListenerContainer container = new RedisMessageListenerContainer();

         container.setConnectionFactory(connectionFactory);

         RedisChannelEnums[] redisChannelEnums = RedisChannelEnums.values();

         if (redisChannelEnums.length > 0) {

             for (RedisChannelEnums redisChannelEnum : redisChannelEnums) {

                 if (redisChannelEnum == null || StringUtils.isEmpty(redisChannelEnum.getCode()) || redisChannelEnum.getClassName()==null) {

                     continue;

                 }

                 //订阅了一个叫pmp和channel 的通道，多通道

                 //一个订阅者接收一个频道信息，新增订阅者需要新增RedisChannelEnums定义+BaseSub的子类

                 String toLowerCase = redisChannelEnum.getClassName().getSimpleName().toLowerCase();

                 BaseSub baseSub = classInstanceMap.get(toLowerCase);

                 container.addMessageListener(listenerAdapter(baseSub), new PatternTopic(redisChannelEnum.getCode()));

             }

         }

         return container;

     }

     /**

      * 配置消息接收处理类

      *

      * @param baseSub

      *         自定义消息接收类

      *

      * @return MessageListenerAdapter

      */

     @Bean()

     @Scope("prototype")

     MessageListenerAdapter listenerAdapter(BaseSub baseSub) {

         //这个地方 是给messageListenerAdapter 传入一个消息接受的处理器，利用反射的方法调用“receiveMessage”

         //也有好几个重载方法，这边默认调用处理器的方法 叫handleMessage 可以自己到源码里面看

         //注意2个通道调用的方法都要为receiveMessage

         return new MessageListenerAdapter(baseSub, "receiveMessage");

     }

 }

@Autowired

public RedisConfig(Map<String, BaseSub> strategyMap) 方法的作用是将所有的配置消息接收处理类注入进来，那么消息接收处理类里面的注解对象也会注入进来。

解决了我们提出的第二个问题

而String toLowerCase = redisChannelEnum.getClassName().getSimpleName().toLowerCase();

BaseSub baseSub = classInstanceMap.get(toLowerCase);

container.addMessageListener(listenerAdapter(baseSub), new PatternTopic(redisChannelEnum.getCode()));

是根据不同的管道对应不同的订阅者，也就是一个订阅者对应一个管道。方便根据不同的业务场景进行处理。

使用这种方式主需要配置redisChannelEnum枚举即可，解决了我们提出的第一个问题。

这样一来，订阅者就变得比较通用了

枚举

RedisChannelEnums作用：定义不同管道对应的订阅者,后期增加一个管道类型只需要增加一个枚举即可

 public enum RedisChannelEnums {

     /**redis频道名称定义 需要与发布者一致*/

     LIVE_INFO_CHANGE("LIVE_INFO_CHANGE", LiveChangeSub.class, "直播信息改变"),

     ;

     /** 枚举定义+描述 */

     private String code;

     private Class<? extends BaseSub> className;

     private String description;

     RedisChannelEnums(String code, Class<? extends BaseSub> className, String description) {

         this.code = code;

         this.className=className;

         this.description = description;

     }

     /** 根据code获取对应的枚举对象 */

     public static RedisChannelEnums getEnum(String code) {

         RedisChannelEnums[] values = RedisChannelEnums.values();

         if (null != code && values.length > 0) {

             for (RedisChannelEnums value : values) {

                 if (value.code == code) {

                     return value;

                 }

             }

         }

         return null;

     }

     /** 该code在枚举列表code属性是否存在 */

     public static boolean containsCode(String code) {

         RedisChannelEnums anEnum = getEnum(code);

         return anEnum != null;

     }

     /** 判断code与枚举中的code是否相同 */

     public static boolean equals(String code, RedisChannelEnums calendarSourceEnum) {

         return calendarSourceEnum.code == code;

     }

     public String getCode() {

         return code;

     }

     public String getDescription() {

         return description;

     }

     public Class<? extends BaseSub> getClassName() {

         return className;

     }

 }

消息模板

BaseSubMessage定义通用的字段，与json字符的通用转换

 @Data

 abstract class BaseSubMessage {

     /** 发布订阅频道名称 */

     private String channel;

     private String extra;

     private String json;

     BaseSubMessage(String json) {

         if(StringUtils.isEmpty(json)){

             return;

         }

         this.json = json;

         Map map = new Gson().fromJson(this.json, Map.class);

         BeanHelper.populate(this, map);

     }

 }

LiveChangeMessage定义当前业务场景的字段

 @Data

 @ToString(callSuper = true)

 public class LiveChangeMessage extends BaseSubMessage {

     /** 直播Ids */

     private String liveIds;

     public LiveChangeMessage(String json) {

         super(json);

     }

 }

订阅者服务

BaseSub定义接收消息的通用方法

 public interface BaseSub {

     /**

      * 接收消息

      * @param jsonMessage  json字符

      */

     void receiveMessage(String jsonMessage);

 }

LiveChangeSub具体消息接收对象

 @Component

 public class LiveChangeSub implements BaseSub {

     /**只是定义的注解测试，可以换成自己的*/

     @Autowired

     private CategoryMapper categoryMapper;

     @Override

     public void receiveMessage(String jsonMessage) {

         System.out.println("项目aries-server.....................");

         //注意通道调用的方法名要和RedisConfig2的listenerAdapter的MessageListenerAdapter参数2相同

         System.out.println("这是 LiveChangeSub" + "-----" + jsonMessage);

         LiveChangeMessage liveChangeMessage = new LiveChangeMessage(jsonMessage);

         System.out.println(liveChangeMessage);

         Category category = categoryMapper.get(1L);

         System.out.println("category:" + category);

     }

 }

总结

发布者配置场景：独立的服务器，独立的项目，A redis缓存服务器

订阅者配置场景：不同于发布者的独立的服务器，独立的项目，A redis缓存服务器

使用场景：一个发布者、一个或者多个订阅者。发布者负责发布消息，订阅者负责接收消息。一旦发布者消息发布出来，那么

订阅者可以通过管道进行监听。同时可以根据不同的管道设置不同的消息接收者或者叫消息处理者。

优点：容易配置，好管理

缺点：由于基于redis去做，不同的redis服务就不适用了。需要考虑消息丢失，持久化的问题。

巴特西

springboot集成redis实现消息发布订阅模式-双通道(跨多服务器)

发布者

枚举定义

消息模板

发布者服务

订阅者

注解配置

枚举

消息模板

订阅者服务

总结

最新文章

热门文章