基于redisson实现发布订阅（多服务间用避坑）

前言

???????今天要分享的是基于Redisson实现信息发布与订阅（以前分享过直接基于redis的实现），如果你是在多服务间基于redisson做信息传递，并且有服务压根就收不到信息，那你一定要看完。
???????今天其实重点是避坑，真正的集成使用就几步。

---

一、redission介绍

???????介绍的文字我都懒得写，其实要我写详细，我也是google，下面直接贴图吧

???????介绍的挺详细的吧，下面还有代码示例哦，不得不说这个GPT插件挺好用的。
???????其实简单理解就一句话：它就是redis的java客户端，做了一层封装。

二、使用步骤

1.引入库

代码如下（示例）：

        <!-- springboot redis集成 -->
        <dependency>
            <groupId>org.springframework.session</groupId>
            <artifactId>spring-session-data-redis</artifactId>
        </dependency>

        <!-- springBoot redisson redis支持 -->
        <dependency>
            <groupId>org.redisson</groupId>
            <artifactId>redisson-spring-boot-starter</artifactId>
            <version>3.24.3</version>
        </dependency>

2.信息发布

/**
 * 告警监听器
 */

import cn.hutool.json.JSONUtil;
import org.redisson.api.RTopic;
import org.redisson.api.RedissonClient;
import org.redisson.codec.SerializationCodec;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import com.xx.xx.alarm.entity.Alarm;
import javax.annotation.PostConstruct;
import javax.annotation.Resource;


@Component
public class AlarmListener{
  
    private static final Logger LOGGER = LoggerFactory.getLogger(AlarmListener.class);

    @Resource
    private RedissonClient redisson;
    public static String WS_ALARM_LISTEN = "WS_ALARM_LISTEN";
    private RTopic topic;



    /**
     * 开启监听
     */
    @PostConstruct
    void openReceiving() {
        topic = redisson.getTopic(WS_ALARM_LISTEN, new SerializationCodec());
    }

	/**
	* 业务需要的地方可以直接待用
	**/
    public void sendNotice(Alarm alarm) {
		//redis 发广播
        try {
            //topic.publish(alarm);
            //屏蔽redisssion监听对class的差异
            String alarmStr = JSONUtil.toJsonStr(alarm);
            topic.publish(alarmStr);
        } catch (Exception e) {
            LOGGER.error("sendNotice失败：", e);
        }
    }

}

???????Alarm是告警实体对象，大家根据自己的业务，可能是其他对象。

---

3、信息订阅

/**
 * 告警监听器
 *
 */

import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.xxxx.entity.Alarm;

import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.StringUtils;
import org.redisson.api.RTopic;
import org.redisson.api.RedissonClient;
import org.redisson.api.listener.MessageListener;
import org.redisson.codec.SerializationCodec;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;

import java.util.List;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.stream.Collectors;

@Component
public class AlarmListener {
   

    private static final Logger LOGGER = LoggerFactory.getLogger(AlarmListener.class);
    
    @Resource
    private RedissonClient redisson;
    public static String WS_ALARM_LISTEN = "WS_ALARM_LISTEN";
    private RTopic topic;




    /**
     * 开启监听
     */
    @PostConstruct
    void openReceiving() {
        topic = redisson.getTopic(WS_ALARM_LISTEN, new SerializationCodec());
        LOGGER.info("监听ws成功：{}", topic);
        
        topic.addListener(String.class, (charSequence, msgStr) -> {
        //TODO 收到消息，去做自己的业务，下面是我们业务的一个示例
            if (StringUtils.isNotEmpty(msgStr) && JSONUtil.isJson(msgStr)) {
                Alarm alarm = JSON.parseObject(msgStr, Alarm.class);
                send(alarm);
            }
        });
    }    
}

???????其实就这么简单，如果是在一个服务里面用，2个监听器是可以合并的。我这里是2个服务里面用。
???????就是因为在2个服务里面用，不知道大家有没有发现topic的publish、addListener的特别之处？可能大家在写的时候，可以直接publish、addListener放入业务对象.class参数。我刚开始也是被坑在这里。一个服务里面publish信息了，另一个服务里死活收不到，用redis-cli去看，发现信息又是放入了主题的，监听的主题也与发布的一致。
???????补充redis-cli命令的使用：

redis-cli -h redis服务ip -p 端口 -a 密码

SUBSCRIBE topic名称

???????我首先想到的是2边版本不一致，于是把新搭建的流水服务的redisson-spring-boot-starter降版本，结果还是一样。
???????然后，就debug，发现监听里的onMessage基类会做如下判断：

???????这个除了判断channel频道，收到的信息，还会判断信息与添加监听addListener时传入的class是否可以转换。

???????网上好多都只提到publish、addListener，但是压根记不会提到传入的class会干嘛，知道问题原因后，我们让数据回归本质，直接用String，这也就形成了我上面的2段。

总结

• 基于redisson实现信息发布订阅就是这么简单几下
• 一定注意publish、addListener不要直接用业务对象（尤其是不在一个服务里，毕竟谁也不能保证对象名一样，也不能保证包路径一样），回归信息的本质用字符串靠谱
• 如果是复杂的信息传递机制，还是用专业的信息中间件
  好了，就写到这里，希望可以帮到大家。