首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >聊聊nacos的DistroConsistencyServiceImpl

聊聊nacos的DistroConsistencyServiceImpl

原创
作者头像
code4it
修改2019-09-11 09:51:56
4350
修改2019-09-11 09:51:56
举报
文章被收录于专栏:码匠的流水账码匠的流水账

本文主要研究一下nacos的DistroConsistencyServiceImpl

ConsistencyService

nacos-1.1.3/naming/src/main/java/com/alibaba/nacos/naming/consistency/ConsistencyService.java

public interface ConsistencyService {
​
    /**
     * Put a data related to a key to Nacos cluster
     *
     * @param key   key of data, this key should be globally unique
     * @param value value of data
     * @throws NacosException
     * @see
     */
    void put(String key, Record value) throws NacosException;
​
    /**
     * Remove a data from Nacos cluster
     *
     * @param key key of data
     * @throws NacosException
     */
    void remove(String key) throws NacosException;
​
    /**
     * Get a data from Nacos cluster
     *
     * @param key key of data
     * @return data related to the key
     * @throws NacosException
     */
    Datum get(String key) throws NacosException;
​
    /**
     * Listen for changes of a data
     *
     * @param key      key of data
     * @param listener callback of data change
     * @throws NacosException
     */
    void listen(String key, RecordListener listener) throws NacosException;
​
    /**
     * Cancel listening of a data
     *
     * @param key      key of data
     * @param listener callback of data change
     * @throws NacosException
     */
    void unlisten(String key, RecordListener listener) throws NacosException;
​
    /**
     * Tell the status of this consistency service
     *
     * @return true if available
     */
    boolean isAvailable();
}
  • ConsistencyService定义了put、remove、get、listen、unlisten、isAvailable方法

EphemeralConsistencyService

nacos-1.1.3/naming/src/main/java/com/alibaba/nacos/naming/consistency/ephemeral/EphemeralConsistencyService.java

public interface EphemeralConsistencyService extends ConsistencyService {
}
  • EphemeralConsistencyService接口继承了ConsistencyService接口

DistroConsistencyServiceImpl

nacos-1.1.3/naming/src/main/java/com/alibaba/nacos/naming/consistency/ephemeral/distro/DistroConsistencyServiceImpl.java

@org.springframework.stereotype.Service("distroConsistencyService")
public class DistroConsistencyServiceImpl implements EphemeralConsistencyService {
​
    private ScheduledExecutorService executor = new ScheduledThreadPoolExecutor(1, new ThreadFactory() {
        @Override
        public Thread newThread(Runnable r) {
            Thread t = new Thread(r);
​
            t.setDaemon(true);
            t.setName("com.alibaba.nacos.naming.distro.notifier");
​
            return t;
        }
    });
​
    @Autowired
    private DistroMapper distroMapper;
​
    @Autowired
    private DataStore dataStore;
​
    @Autowired
    private TaskDispatcher taskDispatcher;
​
    @Autowired
    private DataSyncer dataSyncer;
​
    @Autowired
    private Serializer serializer;
​
    @Autowired
    private ServerListManager serverListManager;
​
    @Autowired
    private SwitchDomain switchDomain;
​
    @Autowired
    private GlobalConfig globalConfig;
​
    private boolean initialized = false;
​
    public volatile Notifier notifier = new Notifier();
​
    private Map<String, CopyOnWriteArrayList<RecordListener>> listeners = new ConcurrentHashMap<>();
​
    private Map<String, String> syncChecksumTasks = new ConcurrentHashMap<>(16);
​
    @PostConstruct
    public void init() {
        GlobalExecutor.submit(new Runnable() {
            @Override
            public void run() {
                try {
                    load();
                } catch (Exception e) {
                    Loggers.DISTRO.error("load data failed.", e);
                }
            }
        });
​
        executor.submit(notifier);
    }
​
    public void load() throws Exception {
        if (SystemUtils.STANDALONE_MODE) {
            initialized = true;
            return;
        }
        // size = 1 means only myself in the list, we need at least one another server alive:
        while (serverListManager.getHealthyServers().size() <= 1) {
            Thread.sleep(1000L);
            Loggers.DISTRO.info("waiting server list init...");
        }
​
        for (Server server : serverListManager.getHealthyServers()) {
            if (NetUtils.localServer().equals(server.getKey())) {
                continue;
            }
            if (Loggers.DISTRO.isDebugEnabled()) {
                Loggers.DISTRO.debug("sync from " + server);
            }
            // try sync data from remote server:
            if (syncAllDataFromRemote(server)) {
                initialized = true;
                return;
            }
        }
    }
​
    //......
​
    public boolean syncAllDataFromRemote(Server server) {
​
        try {
            byte[] data = NamingProxy.getAllData(server.getKey());
            processData(data);
            return true;
        } catch (Exception e) {
            Loggers.DISTRO.error("sync full data from " + server + " failed!", e);
            return false;
        }
    }
​
    public void processData(byte[] data) throws Exception {
        if (data.length > 0) {
            Map<String, Datum<Instances>> datumMap =
                serializer.deserializeMap(data, Instances.class);
​
​
            for (Map.Entry<String, Datum<Instances>> entry : datumMap.entrySet()) {
                dataStore.put(entry.getKey(), entry.getValue());
​
                if (!listeners.containsKey(entry.getKey())) {
                    // pretty sure the service not exist:
                    if (switchDomain.isDefaultInstanceEphemeral()) {
                        // create empty service
                        Loggers.DISTRO.info("creating service {}", entry.getKey());
                        Service service = new Service();
                        String serviceName = KeyBuilder.getServiceName(entry.getKey());
                        String namespaceId = KeyBuilder.getNamespace(entry.getKey());
                        service.setName(serviceName);
                        service.setNamespaceId(namespaceId);
                        service.setGroupName(Constants.DEFAULT_GROUP);
                        // now validate the service. if failed, exception will be thrown
                        service.setLastModifiedMillis(System.currentTimeMillis());
                        service.recalculateChecksum();
                        listeners.get(KeyBuilder.SERVICE_META_KEY_PREFIX).get(0)
                            .onChange(KeyBuilder.buildServiceMetaKey(namespaceId, serviceName), service);
                    }
                }
            }
​
            for (Map.Entry<String, Datum<Instances>> entry : datumMap.entrySet()) {
​
                if (!listeners.containsKey(entry.getKey())) {
                    // Should not happen:
                    Loggers.DISTRO.warn("listener of {} not found.", entry.getKey());
                    continue;
                }
​
                try {
                    for (RecordListener listener : listeners.get(entry.getKey())) {
                        listener.onChange(entry.getKey(), entry.getValue().value);
                    }
                } catch (Exception e) {
                    Loggers.DISTRO.error("[NACOS-DISTRO] error while execute listener of key: {}", entry.getKey(), e);
                    continue;
                }
​
                // Update data store if listener executed successfully:
                dataStore.put(entry.getKey(), entry.getValue());
            }
        }
    }
​
    //......
​
    @Override
    public void put(String key, Record value) throws NacosException {
        onPut(key, value);
        taskDispatcher.addTask(key);
    }
​
    @Override
    public void remove(String key) throws NacosException {
        onRemove(key);
        listeners.remove(key);
    }
​
    @Override
    public Datum get(String key) throws NacosException {
        return dataStore.get(key);
    }
​
    //......
​
    @Override
    public void listen(String key, RecordListener listener) throws NacosException {
        if (!listeners.containsKey(key)) {
            listeners.put(key, new CopyOnWriteArrayList<>());
        }
​
        if (listeners.get(key).contains(listener)) {
            return;
        }
​
        listeners.get(key).add(listener);
    }
​
    @Override
    public void unlisten(String key, RecordListener listener) throws NacosException {
        if (!listeners.containsKey(key)) {
            return;
        }
        for (RecordListener recordListener : listeners.get(key)) {
            if (recordListener.equals(listener)) {
                listeners.get(key).remove(listener);
                break;
            }
        }
    }
​
    @Override
    public boolean isAvailable() {
        return isInitialized() || ServerStatus.UP.name().equals(switchDomain.getOverriddenServerStatus());
    }
​
    //......
}
  • DistroConsistencyServiceImpl实现了EphemeralConsistencyService接口
  • 其init方法会异步执行load方法,该方法会执行syncAllDataFromRemote进行初始化,该方法会通过NamingProxy.getAllData获取data,然后执行processData,它主要是执行回调然后往dataStore添加数据;init方法最后会异步执行Notifier
  • 其put方法会执行onPut方法及taskDispatcher.addTask(key);其remove方法会执行onRemove方法即listeners.remove(key);其get方法直接从dataStore读取;其listen会添加RecordListener;其unlisten则会移除RecordListener;其isAvailable会通过isInitialized及ServerStatus.UP状态来判断

Notifier

nacos-1.1.3/naming/src/main/java/com/alibaba/nacos/naming/consistency/ephemeral/distro/DistroConsistencyServiceImpl.java

    public class Notifier implements Runnable {
​
        private ConcurrentHashMap<String, String> services = new ConcurrentHashMap<>(10 * 1024);
​
        private BlockingQueue<Pair> tasks = new LinkedBlockingQueue<Pair>(1024 * 1024);
​
        public void addTask(String datumKey, ApplyAction action) {
​
            if (services.containsKey(datumKey) && action == ApplyAction.CHANGE) {
                return;
            }
            if (action == ApplyAction.CHANGE) {
                services.put(datumKey, StringUtils.EMPTY);
            }
            tasks.add(Pair.with(datumKey, action));
        }
​
        public int getTaskSize() {
            return tasks.size();
        }
​
        @Override
        public void run() {
            Loggers.DISTRO.info("distro notifier started");
​
            while (true) {
                try {
​
                    Pair pair = tasks.take();
​
                    if (pair == null) {
                        continue;
                    }
​
                    String datumKey = (String) pair.getValue0();
                    ApplyAction action = (ApplyAction) pair.getValue1();
​
                    services.remove(datumKey);
​
                    int count = 0;
​
                    if (!listeners.containsKey(datumKey)) {
                        continue;
                    }
​
                    for (RecordListener listener : listeners.get(datumKey)) {
​
                        count++;
​
                        try {
                            if (action == ApplyAction.CHANGE) {
                                listener.onChange(datumKey, dataStore.get(datumKey).value);
                                continue;
                            }
​
                            if (action == ApplyAction.DELETE) {
                                listener.onDelete(datumKey);
                                continue;
                            }
                        } catch (Throwable e) {
                            Loggers.DISTRO.error("[NACOS-DISTRO] error while notifying listener of key: {}", datumKey, e);
                        }
                    }
​
                    if (Loggers.DISTRO.isDebugEnabled()) {
                        Loggers.DISTRO.debug("[NACOS-DISTRO] datum change notified, key: {}, listener count: {}, action: {}",
                            datumKey, count, action.name());
                    }
                } catch (Throwable e) {
                    Loggers.DISTRO.error("[NACOS-DISTRO] Error while handling notifying task", e);
                }
            }
        }
    }
  • Notifier实现了Runnable接口,其run方法会从LinkedBlockingQueue取task,然后挨个执行listener回调

小结

  • DistroConsistencyServiceImpl实现了EphemeralConsistencyService接口
  • 其init方法会异步执行load方法,该方法会执行syncAllDataFromRemote进行初始化,该方法会通过NamingProxy.getAllData获取data,然后执行processData,它主要是执行回调然后往dataStore添加数据;init方法最后会异步执行Notifier
  • 其put方法会执行onPut方法及taskDispatcher.addTask(key);其remove方法会执行onRemove方法即listeners.remove(key);其get方法直接从dataStore读取;其listen会添加RecordListener;其unlisten则会移除RecordListener;其isAvailable会通过isInitialized及ServerStatus.UP状态来判断

doc

原创声明:本文系作者授权腾讯云开发者社区发表,未经许可,不得转载。

如有侵权,请联系 cloudcommunity@tencent.com 删除。

原创声明:本文系作者授权腾讯云开发者社区发表,未经许可,不得转载。

如有侵权,请联系 cloudcommunity@tencent.com 删除。

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
目录
  • ConsistencyService
  • EphemeralConsistencyService
  • DistroConsistencyServiceImpl
  • Notifier
  • 小结
  • doc
领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档