Nacos配置中心
基于版本1.0.0
Nacos Client
使用Client的Demo代码
public class ClientDemo {
public static void main(String[] args) {
try {
String serverAddr = "127.0.0.1:8848";
String dataId = "nacos-demo.properties";
String group = "DEFAULT_GROUP";
Properties properties = new Properties();
properties.put("serverAddr", serverAddr);
ConfigService configService = NacosFactory.createConfigService(properties);
//没有添加listener 直接获取服务器配置
String content = configService.getConfig(dataId, group, 5000);
System.out.println(content);
} catch (NacosException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
整个过程创建了ConfigService，查看NacosFactory.createConfigService(properties);
public class NacosFactory {
public static ConfigService createConfigService(Properties properties)
throws NacosException {
//查看方法
return ConfigFactory.createConfigService(properties);
}
//.....
}
可以得知通过ConfigFactory工厂类创建ConfigService，继续查看，通过反射创建实例
public class ConfigFactory {
public static ConfigService createConfigService(Properties properties)
throws NacosException {
try {
Class<?> driverImplClass = Class.forName("com.alibaba.nacos.client.config.NacosConfigService");
Constructor constructor = driverImplClass.getConstructor(Properties.class);
//反射创建实例方法
ConfigService vendorImpl =
(ConfigService) constructor.newInstance(properties);
return vendorImpl;
} catch (Throwable e) {
throw new NacosException(-400, e.getMessage());
}
}
}
通过反射形式创建了NacosConfigService实例。ConfigService是个接口，其构造如下
public interface ConfigService {
//获取配置信息
String getConfig(String dataId, String group, long timeoutMs)
throws NacosException;
//添加配置更新的监听器
void addListener(String dataId, String group, Listener listener)
throws NacosException;
//发布配置信息
boolean publishConfig(String dataId, String group, String content)
throws NacosException;
//删除配置
boolean removeConfig(String dataId, String group)
throws NacosException;
//移除监听器
void removeListener(String dataId, String group, Listener listener);
String getServerStatus();
}
NacosConfigService实现如下：
public class NacosConfigService implements ConfigService {
private final long POST_TIMEOUT = 3000L;
private static final String EMPTY = "";
private HttpAgent agent;
private ClientWorker worker;// 负责执行长轮询任务
private String namespace;
private String encode;
private ConfigFilterChainManager configFilterChainManager =
new ConfigFilterChainManager();
public NacosConfigService(Properties properties) throws NacosException {
String encodeTmp = properties.getProperty(PropertyKeyConst.ENCODE);
if (StringUtils.isBlank(encodeTmp)) {
encode = Constants.ENCODE;
} else {
encode = encodeTmp.trim();
}
initNamespace(properties);
//服务信息，用于鉴权、服务地址等的配置信息设置
agent = new MetricsHttpAgent(new ServerHttpAgent(properties));
//启动了ServerHttpAgent，作用是监听服务器列表是否变化。同步方法。具体查看阅读源码了解
agent.start();
//长轮询任务实例，监听配置更新的关键
worker = new ClientWorker(agent, configFilterChainManager);
}
//...
}
ClientWorker源码
public ClientWorker(final HttpAgent agent,
final ConfigFilterChainManager configFilterChainManager) {
this.agent = agent;
this.configFilterChainManager = configFilterChainManager;
executor = Executors.newScheduledThreadPool(1, new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {
Thread t = new Thread(r);
t.setName("com.alibaba.nacos.client.Worker." + agent.getName());
t.setDaemon(true);
return t;
}
});
executorService = Executors.newCachedThreadPool(new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {
Thread t = new Thread(r);
t.setName("com.alibaba.nacos.client.Worker.longPolling" + agent.getName());
t.setDaemon(true);
return t;
}
});
executor.scheduleWithFixedDelay(new Runnable() {
public void run() {
try {
//检查配置是否变化，通过cacheData(linstener)记录配置文件md5信息比较
checkConfigInfo();
} catch (Throwable e) {
}
}
}, 1L, 10L, TimeUnit.MILLISECONDS);//每10毫秒执行一次
}
public void checkConfigInfo() {
// cacheMap保存的 groupKey -> cacheData。groupkey以dataId+group定义
int listenerSize = cacheMap.get().size();
int longingTaskCount = (int)Math.ceil(listenerSize / ParamUtil.getPerTaskConfigSize());
if (longingTaskCount > currentLongingTaskCount) {
for (int i = (int)currentLongingTaskCount; i < longingTaskCount; i++) {
// 执行配置信息更新线程
executorService.execute(new LongPollingRunnable(i));
}
currentLongingTaskCount = longingTaskCount;
}
}
LongPollingRunnable结构
class LongPollingRunnable implements Runnable {
private int taskId;
public LongPollingRunnable(int taskId) {
this.taskId = taskId;
}
public void run() {
try {
List<CacheData> cacheDatas = new ArrayList<CacheData>();
for (CacheData cacheData : cacheMap.get().values()) {
if (cacheData.getTaskId() == taskId) {
cacheDatas.add(cacheData);
try {
// 本地缓存检查
checkLocalConfig(cacheData);
if (cacheData.isUseLocalConfigInfo()) {
// 更新linstener记录的md5值
cacheData.checkListenerMd5();
}
} catch (Exception e) {
}
}
}
List<String> inInitializingCacheList = new ArrayList<String>();
//获取配置中心变化的配置信息，对inInitializingCacheList初始化
List<String> changedGroupKeys =
checkUpdateDataIds(cacheDatas, inInitializingCacheList);
// 配置发生变化则获取服务器最新配置
for (String groupKey : changedGroupKeys) {
String[] key = GroupKey.parseKey(groupKey);
String dataId = key[0];
String group = key[1];
String tenant = null;
if (key.length == 3) {
tenant = key[2];
}
try {
//获取服务器配置信息
String content = getServerConfig(dataId, group, tenant, 3000L);
CacheData cache = cacheMap.get().get(GroupKey.getKeyTenant(dataId, group, tenant));
cache.setContent(content);//设置listener变化配置内容
} catch (NacosException ioe) {
}
}
for (CacheData cacheData : cacheDatas) {
if (!cacheData.isInitializing() || inInitializingCacheList
.contains(GroupKey.getKeyTenant(cacheData.dataId, cacheData.group, cacheData.tenant))) {
//判断监听器md5是否相同，不同更新最新md5
//有变化这发起listener的receiveConfigInfo通知
cacheData.checkListenerMd5();
cacheData.setInitializing(false);
}
}
inInitializingCacheList.clear();
} catch (Throwable e) {
} finally {
//再次执行任务，实现循环
executorService.execute(this);
}
}
}
checkUpdateDataIds代码中调用 checkUpdateConfigStr
/**
* 从Server获取值变化了的DataID列表。返回的对象里只有dataId和group是有效的。 保证不返回NULL。
*/
//probeUpdateString以dataId group md5 tenant命名
List<String> checkUpdateConfigStr(String probeUpdateString,
boolean isInitializingCacheList) {
List<String> params =
Arrays.asList(Constants.PROBE_MODIFY_REQUEST, probeUpdateString);
long timeout = TimeUnit.SECONDS.toMillis(30L);// 30秒超时
List<String> headers = new ArrayList<String>(2);
headers.add("Long-Pulling-Timeout");
headers.add("" + timeout);
// told server do not hang me up if new initializing cacheData added in
// 初始化时直接返回不等待，具体看服务端实现
if (isInitializingCacheList) {
headers.add("Long-Pulling-Timeout-No-Hangup");
headers.add("true");
}
if (StringUtils.isBlank(probeUpdateString)) {
return Collections.emptyList();
}
try {
//调用nacos Api Constants.CONFIG_CONTROLLER_PATH + "/listener 获取服务器配置更新信息
HttpResult result = agent.httpPost(Constants.CONFIG_CONTROLLER_PATH + "/listener", headers, params,
agent.getEncode(), timeout);
if (HttpURLConnection.HTTP_OK == result.code) {
setHealthServer(true);
// 返回有跟新的groupkey
return parseUpdateDataIdResponse(result.content);
} else {
setHealthServer(false);
}
} catch (IOException e) {
}
return Collections.emptyList();
}
checkLocalConfig方法
private void checkLocalConfig(CacheData cacheData) {
final String dataId = cacheData.dataId;
final String group = cacheData.group;
final String tenant = cacheData.tenant;
//获取本地缓存的文件
//LOCAL_SNAPSHOT_PATH = //System.getProperty("JM.SNAPSHOT.PATH",System.getProperty("user.home")) + //File.separator+ "nacos" + File.separator + "config/data"下是否有缓存;
File path = LocalConfigInfoProcessor.getFailoverFile(agent.getName(), dataId, group, tenant);
if (!cacheData.isUseLocalConfigInfo() && path.exists()) {
String content = LocalConfigInfoProcessor.getFailover(agent.getName(), dataId, group, tenant);
String md5 = MD5.getInstance().getMD5String(content);
cacheData.setUseLocalConfigInfo(true);
cacheData.setLocalConfigInfoVersion(path.lastModified());
cacheData.setContent(content);
return;
}
// 有 -> 没有。不通知业务监听器，从server拿到配置后通知。
if (cacheData.isUseLocalConfigInfo() && !path.exists()) {
cacheData.setUseLocalConfigInfo(false);
return;
}
// 有变更
if (cacheData.isUseLocalConfigInfo() && path.exists()
&& cacheData.getLocalConfigInfoVersion() != path.lastModified()) {
String content = LocalConfigInfoProcessor.getFailover(agent.getName(), dataId, group, tenant);
String md5 = MD5.getInstance().getMD5String(content);
cacheData.setUseLocalConfigInfo(true);
cacheData.setLocalConfigInfoVersion(path.lastModified());
cacheData.setContent(content);
return;
}
}
服务端代码导读
API接口/nacos/v1/cs/configs/listener，接口代码
@RequestMapping(value = "/listener", method = RequestMethod.POST)
public void listener(HttpServletRequest request, HttpServletResponse response)
throws ServletException, IOException {
request.setAttribute("org.apache.catalina.ASYNC_SUPPORTED", true);
// client传过来的groupkey
String probeModify = request.getParameter("Listening-Configs");
if (StringUtils.isBlank(probeModify)) {
throw new IllegalArgumentException("invalid probeModify");
}
probeModify = URLDecoder.decode(probeModify, Constants.ENCODE);
Map<String, String> clientMd5Map;
try {
clientMd5Map = MD5Util.getClientMd5Map(probeModify);
} catch (Throwable e) {
throw new IllegalArgumentException("invalid probeModify");
}
// do long-polling
//inner为ConfigServletInner
inner.doPollingConfig(request, response, clientMd5Map, probeModify.length());
}
ConfigServletInner实现
public class ConfigServletInner {
@Autowired
private LongPollingService longPollingService;
@Autowired
private PersistService persistService;
private static final int TRY_GET_LOCK_TIMES = 9;
private static final int START_LONGPOLLING_VERSION_NUM = 204;
/**
* 轮询接口
*/
public String doPollingConfig(HttpServletRequest request, HttpServletResponse response, Map<String, String> clientMd5Map, int probeRequestSize)
throws IOException, ServletException {
// 长轮询
if (LongPollingService.isSupportLongPolling(request)) {
longPollingService.addLongPollingClient(request, response, clientMd5Map, probeRequestSize);
return HttpServletResponse.SC_OK + "";
}
// ...省略短轮询代码
return HttpServletResponse.SC_OK + "";
}
longPollingService.addLongPollingClient
public void addLongPollingClient(HttpServletRequest req, HttpServletResponse rsp, Map<String, String> clientMd5Map,int probeRequestSize) {
String str = req.getHeader(LongPollingService.LONG_POLLING_HEADER);
String noHangUpFlag = req.getHeader(LongPollingService.LONG_POLLING_NO_HANG_UP_HEADER);
String appName = req.getHeader(RequestUtil.CLIENT_APPNAME_HEADER);
String tag = req.getHeader("Vipserver-Tag");
int delayTime = SwitchService.getSwitchInteger(SwitchService.FIXED_DELAY_TIME, 500);
/**
* 提前500ms返回响应，为避免客户端超时 @qiaoyi.dingqy 2013.10.22改动 add delay time for LoadBalance
*/
long timeout = Math.max(10000, Long.parseLong(str) - delayTime);
if (isFixedPolling()) {
timeout = Math.max(10000, getFixedPollingInterval());
} else {
long start = System.currentTimeMillis();
List<String> changedGroups = MD5Util.compareMd5(req, rsp, clientMd5Map);
if (changedGroups.size() > 0) {
generateResponse(req, rsp, changedGroups);
return;
} else if (noHangUpFlag != null && noHangUpFlag.equalsIgnoreCase(TRUE_STR)) {
return;
}
}
String ip = RequestUtil.getRemoteIp(req);
// 一定要由HTTP线程调用，否则离开后容器会立即发送响应
final AsyncContext asyncContext = req.startAsync();
// AsyncContext.setTimeout()的超时时间不准，所以只能自己控制
asyncContext.setTimeout(0L);
// 延迟30s-delayTime执行，再看ClientLongPolling代码
scheduler.execute(
new ClientLongPolling(asyncContext, clientMd5Map, ip, probeRequestSize, timeout, appName, tag));
}
ClientLongPolling代码
ClientLongPolling(AsyncContext ac, Map<String, String> clientMd5Map, String ip, int probeRequestSize,long timeoutTime, String appName, String tag) {
this.asyncContext = ac;
this.clientMd5Map = clientMd5Map;
this.probeRequestSize = probeRequestSize;
this.createTime = System.currentTimeMillis();
this.ip = ip;
this.timeoutTime = timeoutTime;
this.appName = appName;
this.tag = tag;
}
/**
*其中final Queue<ClientLongPolling> allSubs; 是一个队列
* 这个队列维持客户端的请求任务，当调用配置修改api时会触发入队
*
*/
public void run() {
asyncTimeoutFuture = scheduler.schedule(new Runnable() {
public void run() {
try {
getRetainIps().put(ClientLongPolling.this.ip, System.currentTimeMillis());
//删除订阅关系
allSubs.remove(ClientLongPolling.this);
if (isFixedPolling()) {
// 通过比较md5值获取配置变更groupkey
List<String> changedGroups = MD5Util.compareMd5(
(HttpServletRequest)asyncContext.getRequest(),
(HttpServletResponse)asyncContext.getResponse(), clientMd5Map);
if (changedGroups.size() > 0) {
sendResponse(changedGroups);
} else {
sendResponse(null);
}
} else {
sendResponse(null);
}
} catch (Throwable t) {
}
}
}, timeoutTime, TimeUnit.MILLISECONDS);
allSubs.add(this);
}
配置修改API
@RequestMapping(method = RequestMethod.POST)
@ResponseBody
public Boolean publishConfig(...)
throws NacosException {
//...
final Timestamp time = TimeUtils.getCurrentTime();
String betaIps = request.getHeader("betaIps");
ConfigInfo configInfo = new ConfigInfo(dataId, group, tenant, appName, content);
if (StringUtils.isBlank(betaIps)) {
if (StringUtils.isBlank(tag)) {
persistService.insertOrUpdate(srcIp, srcUser, configInfo, time, configAdvanceInfo, false);
// 关键查看这个方法
EventDispatcher.fireEvent(new ConfigDataChangeEvent(false, dataId, group, tenant, time.getTime()));
} else {
persistService.insertOrUpdateTag(configInfo, tag, srcIp, srcUser, time, false);
EventDispatcher.fireEvent(new ConfigDataChangeEvent(false, dataId, group, tenant, tag, time.getTime()));
}
} else { // beta publish
persistService.insertOrUpdateBeta(configInfo, betaIps, srcIp, srcUser, time, false);
EventDispatcher.fireEvent(new ConfigDataChangeEvent(true, dataId, group, tenant, time.getTime()));
}
ConfigTraceService.logPersistenceEvent(dataId, group, tenant, requestIpApp, time.getTime(), LOCAL_IP, ConfigTraceService.PERSISTENCE_EVENT_PUB, content);
return true;
}
EventDispatcher.fireEvent
static public void fireEvent(Event event) {
if (null == event) {
throw new IllegalArgumentException();
}
//从linstener中获取变更事件触发实例
for (AbstractEventListener listener : getEntry(event.getClass()).listeners) {
try {
listener.onEvent(event);
} catch (Exception e) {
log.error(e.toString(), e);
}
}
}
而listeners维护在Entry中，Entry重LISTENER_HUB中获取，而LISTENER_HUB则通过addEventListener添加linstener
// AbstractEventListener的构造方法会触发 EventDispatcher.addEventListener。 因此查看
//AbstractEventListener的实现
static public abstract class AbstractEventListener {
public AbstractEventListener() {
EventDispatcher.addEventListener(this);
}
}
查看LongPollingService,当客户端添加了linstener时allSub维护的订阅关系，而配置变更时会触发onEvent事件,onEvent执行DataChangeTask任务。
@Override
public void onEvent(Event event) {
if (isFixedPolling()) {
// ignore
} else {
if (event instanceof LocalDataChangeEvent) {
LocalDataChangeEvent evt = (LocalDataChangeEvent)event;
scheduler.execute(new DataChangeTask(evt.groupKey, evt.isBeta, evt.betaIps));
}
}
}
DataChangeTask 实现
public void run() {
try {
ConfigService.getContentBetaMd5(groupKey);
for (Iterator<ClientLongPolling> iter = allSubs.iterator(); iter.hasNext(); ) {
ClientLongPolling clientSub = iter.next();
//找到客户端订阅的groupkey任务
if (clientSub.clientMd5Map.containsKey(groupKey)) {
// ...
getRetainIps().put(clientSub.ip, System.currentTimeMillis());
iter.remove(); // 删除订阅关系
//通过长轮询，返回变更消息通知客户端
clientSub.sendResponse(Arrays.asList(groupKey));
}
}
} catch (Throwable t) {
}
}
网友评论