Nacos 致力于帮助您发现、配置和管理微服务

/* * Copyright 1999-2018 Alibaba Group Holding Ltd. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package com.alibaba.nacos.client.config.impl; import com.alibaba.nacos.api.PropertyKeyConst; import com.alibaba.nacos.api.common.Constants; import com.alibaba.nacos.api.config.ConfigType; import com.alibaba.nacos.api.config.listener.Listener; import com.alibaba.nacos.api.config.remote.request.ClientConfigMetricRequest; import com.alibaba.nacos.api.config.remote.request.ConfigBatchListenRequest; import com.alibaba.nacos.api.config.remote.request.ConfigChangeNotifyRequest; import com.alibaba.nacos.api.config.remote.request.ConfigPublishRequest; import com.alibaba.nacos.api.config.remote.request.ConfigQueryRequest; import com.alibaba.nacos.api.config.remote.request.ConfigRemoveRequest; import com.alibaba.nacos.api.config.remote.response.ClientConfigMetricResponse; import com.alibaba.nacos.api.config.remote.response.ConfigChangeBatchListenResponse; import com.alibaba.nacos.api.config.remote.response.ConfigChangeNotifyResponse; import com.alibaba.nacos.api.config.remote.response.ConfigPublishResponse; import com.alibaba.nacos.api.config.remote.response.ConfigQueryResponse; import com.alibaba.nacos.api.config.remote.response.ConfigRemoveResponse; import com.alibaba.nacos.api.exception.NacosException; import com.alibaba.nacos.api.remote.RemoteConstants; import com.alibaba.nacos.api.remote.request.Request; import com.alibaba.nacos.api.remote.response.Response; import com.alibaba.nacos.client.config.common.GroupKey; import com.alibaba.nacos.client.config.filter.impl.ConfigFilterChainManager; import com.alibaba.nacos.client.config.filter.impl.ConfigResponse; import com.alibaba.nacos.client.config.utils.ContentUtils; import com.alibaba.nacos.client.env.NacosClientProperties; import com.alibaba.nacos.client.monitor.MetricsMonitor; import com.alibaba.nacos.client.naming.utils.CollectionUtils; import com.alibaba.nacos.client.utils.AppNameUtils; import com.alibaba.nacos.client.utils.EnvUtil; import com.alibaba.nacos.client.utils.LogUtils; import com.alibaba.nacos.client.utils.ParamUtil; import com.alibaba.nacos.client.utils.TenantUtil; import com.alibaba.nacos.common.lifecycle.Closeable; import com.alibaba.nacos.common.notify.Event; import com.alibaba.nacos.common.notify.NotifyCenter; import com.alibaba.nacos.common.notify.listener.Subscriber; import com.alibaba.nacos.common.remote.ConnectionType; import com.alibaba.nacos.common.remote.client.Connection; import com.alibaba.nacos.common.remote.client.ConnectionEventListener; import com.alibaba.nacos.common.remote.client.RpcClient; import com.alibaba.nacos.common.remote.client.RpcClientFactory; import com.alibaba.nacos.common.remote.client.RpcClientTlsConfig; import com.alibaba.nacos.common.remote.client.ServerListFactory; import com.alibaba.nacos.common.utils.ConvertUtils; import com.alibaba.nacos.common.utils.JacksonUtils; import com.alibaba.nacos.common.utils.MD5Utils; import com.alibaba.nacos.common.utils.StringUtils; import com.alibaba.nacos.common.utils.ThreadUtils; import com.alibaba.nacos.common.utils.VersionUtils; import com.alibaba.nacos.plugin.auth.api.RequestResource; import com.google.gson.Gson; import com.google.gson.JsonObject; import org.slf4j.Logger; import java.io.File; import java.util.ArrayList; import java.util.Collection; import java.util.HashMap; import java.util.HashSet; import java.util.Iterator; import java.util.LinkedList; import java.util.List; import java.util.Map; import java.util.Set; import java.util.UUID; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.BlockingQueue; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; import static com.alibaba.nacos.api.common.Constants.ENCODE; /** * Long polling. * * @author Nacos */ public class ClientWorker implements Closeable { private static final Logger LOGGER = LogUtils.logger(ClientWorker.class); private static final String NOTIFY_HEADER = "notify"; private static final String TAG_PARAM = "tag"; private static final String APP_NAME_PARAM = "appName"; private static final String BETAIPS_PARAM = "betaIps"; private static final String TYPE_PARAM = "type"; private static final String ENCRYPTED_DATA_KEY_PARAM = "encryptedDataKey"; /** * groupKey -> cacheData. */ private final AtomicReference<Map<String, CacheData>> cacheMap = new AtomicReference<>(new HashMap<>()); private final ConfigFilterChainManager configFilterChainManager; private final String uuid = UUID.randomUUID().toString(); private long timeout; private final ConfigRpcTransportClient agent; private int taskPenaltyTime; private boolean enableRemoteSyncConfig = false; private static final int MIN_THREAD_NUM = 2; private static final int THREAD_MULTIPLE = 1; /** * index(taskId)-> total cache count for this taskId. */ private final List<AtomicInteger> taskIdCacheCountList = new ArrayList<>(); /** * Add listeners for data. * * @param dataId dataId of data * @param group group of data * @param listeners listeners */ public void addListeners(String dataId, String group, List<? extends Listener> listeners) throws NacosException { group = blank2defaultGroup(group); CacheData cache = addCacheDataIfAbsent(dataId, group); synchronized (cache) { for (Listener listener : listeners) { cache.addListener(listener); } cache.setDiscard(false); cache.setConsistentWithServer(false); agent.notifyListenConfig(); } } /** * Add listeners for tenant. * * @param dataId dataId of data * @param group group of data * @param listeners listeners * @throws NacosException nacos exception */ public void addTenantListeners(String dataId, String group, List<? extends Listener> listeners) throws NacosException { group = blank2defaultGroup(group); String tenant = agent.getTenant(); CacheData cache = addCacheDataIfAbsent(dataId, group, tenant); synchronized (cache) { for (Listener listener : listeners) { cache.addListener(listener); } cache.setDiscard(false); cache.setConsistentWithServer(false); agent.notifyListenConfig(); } } /** * Add listeners for tenant with content. * * @param dataId dataId of data * @param group group of data * @param content content * @param encryptedDataKey encryptedDataKey * @param listeners listeners * @throws NacosException nacos exception */ public void addTenantListenersWithContent(String dataId, String group, String content, String encryptedDataKey, List<? extends Listener> listeners) throws