nacos中DistroConsistencyServiceImpl的原理和作用是什么

本篇内容介绍了“nacos中DistroConsistencyServiceImpl的原理和作用是什么”的有关知识，在实际案例的操作过程中，不少人都会遇到这样的困境，接下来就让小编带领大家学习一下如何处理这些情况吧！希望大家仔细阅读，能够学有所成！

创新互联建站服务项目包括昌图网站建设、昌图网站制作、昌图网页制作以及昌图网络营销策划等。多年来，我们专注于互联网行业，利用自身积累的技术优势、行业经验、深度合作伙伴关系等，向广大中小型企业、政府机构等提供互联网行业的解决方案，昌图网站推广取得了明显的社会效益与经济效益。目前，我们服务的客户以成都为中心已经辐射到昌图省份的部分城市，未来相信会继续扩大服务区域并继续获得客户的支持与信任！

序

本文主要研究一下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状态来判断

“nacos中DistroConsistencyServiceImpl的原理和作用是什么”的内容就介绍到这里了，感谢大家的阅读。如果想了解更多行业相关的知识可以关注创新互联网站，小编将为大家输出更多高质量的实用文章！

网站标题：nacos中DistroConsistencyServiceImpl的原理和作用是什么
分享链接：https://www.cdcxhl.com/article44/jhgdhe.html

成都网站建设公司_创新互联，为您提供定制网站、品牌网站制作、品牌网站设计、建站公司、网站建设、网站导航

广告

声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源： 创新互联

猜你还喜欢下面的内容

小程序开发

手机网站建设知识

分类信息网