我在生产项目里是如何使用Redis发布订阅的？（二）Java版代码实现（含源码）

public interface ICacheUpdate {
    public void update();
}

public class InfoService implements ICacheUpdate {
  private static Logger logger = LoggerFactory.getLogger(InfoService.class);
  @Autowired
  private RedisCache redisCache;
  @Autowired
  private InfoMapper infoMapper;
  /**
   * 按信息类型分类查询信息
   * @return
   */
  public Map>> selectAllInfo(){
    Map>> resultMap = new HashMap>>();
    List infoTypeList = infoMapper.selectInfoType();//信息表中所有涉及的信息类型
    logger.info("-------按信息类型查找公共信息开始----"+infoTypeList);
    if(infoTypeList!=null && infoTypeList.size()>0) {
      for (String infoType : infoTypeList) {
        List

public class RedisMsgPubSub extends JedisPubSub {
    private static Logger logger = LoggerFactory.getLogger(RedisMsgPubSub.class);
    private Map updates = new HashMap();
    //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...
}

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!");
            }
        }
    }
}

/**
 * 加载系统参数
 */
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...
    }
}

com.xxx.listener.CacheInitListener