你真的了解Redis的发布订阅？

Redis是我们很常用的一款nosql数据库产品，我们通常会用 Redis 来配合关系型数据库一起使用，弥补关系型数据库的不足。

其中，Redis的发布订阅功能也是它的一大亮点。虽然它不是一款专门做发布订阅的产品，但其自带的发布订阅功能已经满足我们日常需求。

那Redis的发布订阅功能的原理和它都可以用在哪些场景呢？今天我们就来探讨一下这个问题。

什么是发布订阅

所谓发布订阅，就是消息发布者发布消息及消息订阅者接收消息，二者通过某种媒介关联起来。这类似以前的『订报』，当我们订阅了 某种报纸 后（比如财经报），每当报纸有新的期刊出版后，就会有邮递员给我们送过来。即，只有定了这种报纸才会收到出版社发布的这种新报纸。

Redis的发布订阅功能也是类似，首先要有消息的发布者，其次要有消息的订阅者。有了消息发布者和订阅者之后，还缺少什么？

那就是上述的『某种报纸』，并不是出版社出版的每一种报纸（如人民日报，财经报，体育报）都给你送过来，而是明确你要定哪一种，你定了哪一种才给你送哪一种。

回到Redis的发布订阅上，上述的『某种报纸』就抽象为频道 channel ，客户端订阅了某 channel 后，当发布者通过此 channel 发布消息时，所有订阅者就会收到该频道发布的消息。

发布和订阅机制

当一个客户端通过 PUBLISH 命令向订阅者发送信息的时候，我们称这个客户端为发布者(publisher)。

而当一个客户端使用 SUBSCRIBE 或者 PSUBSCRIBE命令接收信息的时候，我们称这个客户端为订阅者(subscriber)。

为了解耦发布者(publisher)和订阅者(subscriber)之间的关系，Redis 使用了 channel (频道)作为两者的中介 —— 发布者将信息直接发布给 channel ，而 channel 负责将信息发送给适当的订阅者，发布者和订阅者之间没有相互关系，也不知道对方的存在。

就会收到该消息。

原理

Redis是使用C实现的，通过分析 Redis 源码里的 pubsub.c 文件，了解发布和订阅机制的底层实现，籍此加深对 Redis 的理解。

Redis 通过 PUBLISH 、SUBSCRIBE 和 PSUBSCRIBE 等命令实现发布和订阅功能。

通过 SUBSCRIBE 命令订阅某频道后，redis-server 里维护了一个字典，字典的键就是一个个 channel ，而字典的值则是一个链表，链表中保存了所有订阅这个 channel 的客户端。SUBSCRIBE 命令的关键，就是将客户端添加到给定 channel 的订阅链表中。

通过 PUBLISH 命令向订阅者发送消息，redis-server 会使用给定的频道作为键，在它所维护的 channel 字典中查找记录了订阅这个频道的所有客户端的链表，遍历这个链表，将消息发布给所有订阅者。

详细参考： Redis 发布/订阅机制原理分析

业务场景

明确了Redis发布订阅的原理和基本流程后，我们来看一下Redis的发布订阅到底具体能做什么。

1、异步消息通知

比如渠道在调支付平台的时候，我们可以用回调的方式给支付平台一个我们的回调接口来通知我们支付状态，还可以利用Redis的发布订阅来实现。比如我们发起支付的同时订阅频道 pay_notice_ + wk (假如我们的渠道标识是wk，不能让其他渠道也订阅这个频道)，当支付平台处理完成后，支付平台往该频道发布消息，告诉频道的订阅者该订单的支付信息及状态。收到消息后，根据消息内容更新订单信息及后续操作。

当很多人都调用支付平台时，支付时都去订阅同一个频道会有问题。比如用户A支付完订阅频道 pay_notice_wk ，在支付平台未处理完时，用户B支付完也订阅了 pay_notice_wk ，当A收到通知后，接着B的支付通知也发布了，这时渠道收不到第二次消息发布。因为同一个频道收到消息后，订阅自动取消，也就是订阅是一次性的。

所以我们订阅的订单支付状态的频道就得唯一，一个订单一个频道，我们可以在频道上加上订单号 pay_notice_wk +orderNo保证频道唯一。这样我们可以把频道号在支付时当做参数一并传过去，支付平台处理完就可以用此频道发布消息给我们了。（实际大多使用接口回调通知的方式，因为用Redis发布订阅限制条件苛刻，系统间必须共用一套Redis）

2、任务通知

比如通过跑批系统通知应用系统做一些事（跑批系统无法拿到用户数据，且应用系统又不能做定时任务的情况下）。如每天凌晨3点提前加载一些用户的用户数据到Redis，应用系统不能做定时任务，可以通过系统公共的Redis来由跑批系统发布任务给应用系统，应用系统收到指令，去做相应的操作。

这里需要注意的是在线上集群部署的情况下，所有服务实例都会收到通知，都要做同样的操作吗？完全没必要。可以用Redis实现锁机制，其中一台实例拿到锁后执行任务。另外如果任务比较耗时，可以不用锁，可以考虑一下任务分片执行。当然这不在本文的讨论范畴，这里不在赘述。

3、参数刷新加载

众所周知，我们用Redis无非就是将系统中不怎么变的、查询又比较频繁的数据缓存起来，例如我们系统首页的轮播图啊，页面的动态链接啊，一些系统参数啊，公共数据啊都加载到Redis，然后有个后台管理系统去配置修改这些数据。

打个比方我们首页的轮播图要再增加一个图，那我们就在后管系统加上，加上就完事了吗？当然没有，因为Redis里还是老数据。那你会说不是有过期时间吗？是的，但有的过期时间设置的较长如24小时并且我们想立即生效怎么办？这时候我们就可以利用Redis的发布订阅机制来实现数据的实时刷新。当我们修改完数据后，点击刷新按钮，通过发布订阅机制，订阅者接收到消息后调用重新加载的方法即可。

代码实现

发布订阅的理论以及使用场景大家都已经有了大致了解了，但是怎么用代码实现发布订阅呢？在这里给大家分享一下实现方式。

我们以第三种使用场景为例，先来看一下整体实现类图吧。

解释一下，这里我们首先定义一个统一接口 ICacheUpdate ，只有一个 update 方法，我们令 Service 层实现这个方法，执行具体的更新操作。我们再来看 RedisMsgPubSub ，它继承 redis.clients.jedis.JedisPubSub ，主要重写其 onMessage() 方法（订阅的频道有消息到来时会触发这个方法），我们在这个方法里调用 RedisMsgPubSub 的 update 方法执行更新操作。当我们有多个 Service 实现 ICacheUpdate 时，我们就非常迫切地需要一个管理器来集中管理这些 Service ，并且当触发onMessage方法时要告诉onMessage方法具体调用哪个 ICacheUpdate 的实现类，所以我们有了 PubSubManager 。并且我们单独开启一个线程来维护发布订阅，所以管理器继承了 Thread 类。

具体代码：

统一接口

ICacheUpdate.java

public interface ICacheUpdate {
    public void update();
}
复制代码

Service层

实现ICacheUpdate的update方法，执行具体的更新操作

InfoService.java

public class InfoService implements ICacheUpdate {
	private static Logger logger = LoggerFactory.getLogger(InfoService.class);
	@Autowired
	private RedisCache redisCache;
	@Autowired
	private InfoMapper infoMapper;
	/**
	 * 按信息类型分类查询信息
	 * @return
	 */
	public Map<String, List<Map<String, Object>>> selectAllInfo(){
		Map<String, List<Map<String, Object>>> resultMap = new HashMap<String, List<Map<String, Object>>>();
		List<String> infoTypeList = infoMapper.selectInfoType();//信息表中所有涉及的信息类型
		logger.info("-------按信息类型查找公共信息开始----"+infoTypeList);
		if(infoTypeList!=null && infoTypeList.size()>0) {
			for (String infoType : infoTypeList) {
				List<Map<String, Object>> result = infoMapper.selectByInfoType(infoType);
				resultMap.put(infoType, result);
			}
		}
		return resultMap;
	}
	@Override
	public void update() {
		//缓存首页信息
		logger.info("InfoService selectAllInfo 刷新缓存");
		Map<String, List<Map<String, Object>>> resultMap = this.selectAllInfo();
		Set<String> keySet = resultMap.keySet();
		for(String key:keySet){
			List<Map<String, Object>> value = resultMap.get(key);
			redisCache.putObject(GlobalSt.PUBLIC_INFO_ALL+key, value);
		}
	}
}
复制代码

Redis发布订阅的扩展类

作用：

1、统一管理ICacheUpdate，把所有实现ICacheUpdate接口的类添加到updates容器

2、重写onMessage方法，订阅到消息后进行刷新缓存的操作

RedisMsgPubSub.java

/**
 * Redis发布订阅的扩展类
 * 作用：1、统一管理ICacheUpdate，把所有实现ICacheUpdate接口的类添加到updates容器
 * 2、重写onMessage方法，订阅到消息后进行刷新缓存的操作
 */
public class RedisMsgPubSub extends JedisPubSub {
    private static Logger logger = LoggerFactory.getLogger(RedisMsgPubSub.class);
    private Map<String , ICacheUpdate> updates = new HashMap<String , ICacheUpdate>();
    //1、由updates统一管理ICacheUpdate
    public boolean addListener(String key , ICacheUpdate update) {
        if(update == null) 
            return false;
	updates.put(key, update);
	return true;
    }
    /**
     * 2、重写onMessage方法，订阅到消息后进行刷新缓存的操作
     * 订阅频道收到的消息
     */
    @Override  
    public void onMessage(String channel, String message) {
        logger.info("RedisMsgPubSub onMessage channel:{},message :{}" ,channel, message);
        ICacheUpdate updater = null;
        if(StringUtil.isNotEmpty(message)) 
            updater = updates.get(message);
        if(updater!=null)
            updater.update();
    }
    //other code...
}
复制代码

发布订阅的管理器

执行的操作：

1、将所有需要刷新加载的Service类（实现ICacheUpdate接口）添加到RedisMsgPubSub的updates中

2、启动线程订阅pubsub_config频道，收到消息后的五秒后再次订阅（避免订阅到一次消息后结束订阅）

PubSubManager.java

public class PubSubManager extends Thread{
    private static Logger logger = LoggerFactory.getLogger(PubSubManager.class);

    public static Jedis jedis;
    RedisMsgPubSub msgPubSub = new RedisMsgPubSub();
    //频道
    public static final String PUNSUB_CONFIG = "pubsub_config";
    //1.将所有需要刷新加载的Service类（实现ICacheUpdate接口）添加到RedisMsgPubSub的updates中
    public boolean addListener(String key, ICacheUpdate listener){
        return msgPubSub.addListener(key,listener);
    }
    @Override
    public void run(){
        while (true){
            try {
                JedisPool jedisPool = SpringTools.getBean("jedisPool", JedisPool.class);
                if(jedisPool!=null){
                    jedis = jedisPool.getResource();
                    if(jedis!=null){
                        //2.启动线程订阅pubsub_config频道 阻塞
                        jedis.subscribe(msgPubSub,PUNSUB_CONFIG);
                    }
                }
            } catch (Exception e) {
                logger.error("redis connect error!");
            } finally {
                if(jedis!=null)
                    jedis.close();
            }
            try {
                //3.收到消息后的五秒后再次订阅（避免订阅到一次消息后结束订阅）
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                logger.error("InterruptedException in redis sleep!");
            }
        }
    }
}
复制代码

到此，Redis的发布订阅大致已经实现。我们什么时候启用呢？我们可以选择在启动项目时完成订阅和基础数据的加载，所以我们通过实现 javax.servlet.SevletContextListener 来完成这一操作。然后将监听器添加到 web.xml 。

CacheInitListener.java

/**
 * 加载系统参数
 */
public class CacheInitListener implements ServletContextListener{
    private static Logger logger = LoggerFactory.getLogger(CacheInitListener.class);

    @Override
    public void contextDestroyed(ServletContextEvent arg0) {
    }

    @Override
    public void contextInitialized(ServletContextEvent arg0) {
        logger.info("---CacheListener初始化开始---");
        init();
        logger.info("---CacheListener初始化结束---");
    }

    public void init() {
        try {
            //获得管理器
            PubSubManager pubSubManager = SpringTools.getBean("pubSubManager", PubSubManager.class);

            InfoService infoService = SpringTools.getBean("infoService", InfoService.class);
            //添加到管理器
            pubSubManager.addListener("infoService", infoService);
            //other service...

            //启动线程执行订阅操作
            pubSubManager.start();
            //初始化加载
            loadParamToRedis();
        } catch (Exception e) {
            logger.info(e.getMessage(), e);
        }
    }

    private void loadParamToRedis() {
        InfoService infoService = SpringTools.getBean("infoService", InfoService.class);
        infoService.update();
        //other service...
    }
}
复制代码

web.xml

<listener>
	<listener-class>com.xxx.listener.CacheInitListener</listener-class>
</listener>
复制代码

【end】

文章首发于公众号@编程大道