目录

背景

Redis发布订阅

MQ广播消息

配置中心Nacos，Zookeeper监听

注册中心获取服务节点ip端口接口调用

本地定时任务兜底

---

背景

我们在系统开发过程中，为了减少数据库和redis缓存的查询以提升接口性能，有时候会把一些常用的、变动不是很频繁的数据放到本地缓存，因为我们的服务是多节点部署的，在缓存的数据发生变动时，怎么快速的将整个集群中的各个节点上的数据同步更新呢？

这里总结了如下几种常用的解决方案：

• Redis发布订阅
• MQ广播消息
• 配置中心nacos，zookeeper监听
• 注册中心获取服务节点ip端口进行接口调用
• 定时任务兜底

        下面介绍一下几种方案的实现思路，同时也给出了主要的实现代码，具体的刷新逻辑可以根据自己的需要进行实现。

Redis发布订阅

        借助redis的发布订阅机制，服务使用Redis注册一个channel监听，当有数据变更的时候往channel发布一个消息，这样集群中的各个节点都会收到这个消息执行本地缓存的刷新操作。

        测试代码如下所示：

@Resource
RedisTemplate redisTemplate;

@Test
public void test() throws InterruptedException {
	new Thread(()->refreshCacheListener()).start();
	new Thread(()->refreshCacheListener()).start();
	Thread.sleep(1000L);
	redisTemplate.convertAndSend("refreshCache","刷新本地缓存了");
	Thread.sleep(1000L);
	redisTemplate.convertAndSend("refreshCache","又刷新了");
}

private void refreshCacheListener() {
	RedisConnection connection = redisTemplate.getConnectionFactory().getConnection();
	connection.subscribe((message,pattern)->{
		log.info("刷新消息:thread:{},message:{}",Thread.currentThread().getId(),new String(message.getBody()));
		//TODO 根据收到的消息执行刷新本地缓存的逻辑
	},"refreshCache".getBytes());
}

运行输出如下：

刷新消息:thread:110,message:"刷新本地缓存了"
刷新消息:thread:111,message:"刷新本地缓存了"
刷新消息:thread:111,message:"又刷新了"
刷新消息:thread:110,message:"又刷新了"

MQ广播消息

        借助MQ的广播消息的机制，当缓存数据变更的时候发布一个广播消息，集群中的各个节点都会收到这个消息执行本地缓存的刷新操作。

        以RabbitMQ实现广播消息举例：通过rabbitmq的fanout广播类型交换机，然后程序中会声明一个临时队列绑定到广播交换机上，这样每个机器都会声明一个不同名字的临时队列绑定到广播交换机上，往广播交换机上发送一个消息，所有的机器都会消费到该消息，临时队列不会持久化，服务重启或者停止后，原先的临时队列会被删除，然后重新创建新的临时队列进行绑定。 

        RabbitMQ广播消息代码示例如下 

@Slf4j
@Component
public class RefreshCacheConsumer {
 
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(), //注意这里不要定义队列名称,系统会随机产生
            exchange = @Exchange(value = "refreshCacheExchange",type = ExchangeTypes.FANOUT)
    ))
    public void process(String payload) {
        log.info("刷新广播消息receive:{}",payload)
        //TODO 根据收到的消息执行刷新本地缓存的逻辑
    }
}

配置中心Nacos，Zookeeper监听

Nacos配置中心 

        借助Nacos配置中心的监听机制实现本地缓存刷新的原理也和上面类似，服务监听一个配置中心文件的内容变更，当缓存数据变更时发布新的配置内容，大致代码如下所示

@Slf4j
@Component
public class CacheRefresher implements InitializingBean {
    @Resource
    private NacosConfigManager nacosConfigManager;
 
    @Override
    public void afterPropertiesSet() throws Exception {
        nacosConfigManager.getConfigService().addListener("nacos-refresh-cache.properties", "DEFAULT_GROUP",
            new Listener() {
 
                @Override
                public Executor getExecutor() {
                    return Executors.newSingleThreadExecutor();
                }
 
                @Override
                public void receiveConfigInfo(String configInfo) {
                    log.info("刷新缓存recieve:{}", configInfo);
					//TODO 根据收到的消息执行刷新本地缓存的逻辑
                }
        });
    }
}


public class ConfigServerDemo {

    public static void main(String[] args) throws Exception {
        String serverAddr = "localhost";
        String dataId = "nacos-refresh-cache.properties";
        String group = "DEFAULT_GROUP";
        Properties properties = new Properties();
        properties.put(PropertyKeyConst.SERVER_ADDR, serverAddr);
        ConfigService configService = NacosFactory.createConfigService(properties);
       
        //缓存变更,发布配置
        configService.publishConfig(dataId,group,"refreshCache");

        Thread.sleep(3000);
        content = configService.getConfig(dataId, group, 5000);
        System.out.println(content);
    }
}

Zookeeper节点监听 

        利用zookeeper的watch机制，服务监听节点的数据变更事件，在事件中调用刷新缓存的逻辑，当有缓存数据变更时更新zookeeper对应节点的数据即可。关于zookeeper的使用如果不太了解的话，可以看下我前面的一篇介绍zookeeper的curator客户端使用的博客 (8)zookeeper开源客户端Curator使用介绍

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.NodeCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
 
public class TestNodeCacheListener {
 
	/** zookeeper地址 */
	static final String CONNECT_ADDR = "192.168.74.4:2181,192.168.74.5:2181,192.168.74.6:2181";
	/** session超时时间 */
	static final int SESSION_OUTTIME = 5000;//ms 
	
	@SuppressWarnings("resource")
	public static void main(String[] args) throws Exception {
		//1 重试策略：初试时间为1s 重试10次
		RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 10);
		CuratorFramework client = CuratorFrameworkFactory.builder()
				.connectString(CONNECT_ADDR)
				.connectionTimeoutMs(SESSION_OUTTIME)
				.retryPolicy(retryPolicy)
				.build();
		client.start();
		
		//dataIsCompressed:true/false 表示是否对节点数据进行压缩
		final NodeCache cache = new NodeCache(client, "/super", false);
		//start(true) 里面的true表示对节点进行初始化
		cache.start(false);
		
		cache.getListenable().addListener(new NodeCacheListener() {
			
			@Override
			public void nodeChanged() throws Exception {
				ChildData data = cache.getCurrentData(); 
				if (null != data) { 
					System.out.println("路径为：" + cache.getCurrentData().getPath());
					System.out.println("数据为：" + new String(cache.getCurrentData().getData()));
					System.out.println("状态为：" + cache.getCurrentData().getStat());
					System.out.println("---------------------------------------");
				} else { 
					System.out.println("节点被删除!"); 
				}
 
			}
		});
		
		Thread.sleep(1000);
		client.create().forPath("/super", "123".getBytes());
		
		Thread.sleep(1000);
		client.setData().forPath("/super", "456".getBytes());
		
		Thread.sleep(1000);
		client.delete().forPath("/super");
		
		Thread.sleep(Integer.MAX_VALUE);
	}
}

        程序输出如下所示： 

路径为：/super
数据为：123
状态为：429496729934,429496729934,1542033727201,1542033727201,0,0,0,0,3,0,429496729934
 
---------------------------------------
路径为：/super
数据为：456
状态为：429496729934,429496729935,1542033727201,1542033728271,1,0,0,0,3,0,429496729934
 
---------------------------------------
节点被删除!

注册中心获取服务节点ip端口接口调用

        上面的几个方案都是异步更新的，无法立即同步获取缓存刷新的结果。我们的服务都是集群部署的，可以提供一个缓存刷新的接口，缓存变更时获取集群所有节点的ip和端口列表，挨个调用缓存刷新接口，这样我们就可以同步获取到缓存刷新的结果了，如果刷新失败可以进行重试操作。

        我们项目开发中一般都会使用注册中心，通过注册中心可以获取服务的所有节点的ip和端口列表，我们以Nacos为例，大致代码如下所示

@Resource
NacosRegistration registration;
@Resource
RestTemplate restTemplate;

@Test
public void test2() throws Exception {
	//变更的缓存key
	String merchantNo="T001";
	NamingService namingService = registration.getNacosNamingService();
	List<Instance> instanceList = namingService.getAllInstances("pay-service");
	if(CollectionUtils.isNotEmpty(instanceList)){
		//遍历集群中的服务列表,挨个调用接口进行缓存刷新
		for(Instance instance:instanceList){
			String url = new StringBuilder()
					.append("http://")
					.append(instance.getIp())
					.append(":")
					.append(instance.getPort())
					.append("/refreshCache?merchantNo=")
					.append(merchantNo).toString();
			log.info("当前服务节点,url:{}",url);
			ResponseEntity<Response> entity = restTemplate.getForEntity(url, Response.class);
			Response response = entity.getBody();
		}
	}
}

本地定时任务兜底

        上面几种解决方法或多或少都有失败的可能性，可以通过定时任务刷新本地缓存进行兜底。

        定时任务可以通过spring schedule、Timer、quartz等方案进行实现。

        spring schedule定时任务大致代码如下所示：

/**
 * 初始延迟5分钟，每执行完一次后，固定间隔10分钟执行一次<br/>
 * 
 */
@Scheduled(initialDelay = 1000*60*5,fixedDelay = 1000*60*10)
@Override
public Response refreshCacheTask() {
	log.info("刷新缓存信息start===============================");
	
	log.info("刷新缓存信息信息end================================="));
	return Response.success();
}

        我所了解的几种常见的刷新本地缓存的方案介绍完了， 如果还有新的方案，欢迎评论区补充哈。