上一篇文章，我们简单的看了一下diamond server的功能及所做的事情，本篇我们看一下diamond client 所做的事情。
首先我们说明的是，应用系统想要获得diamond中的动态配置，需要设置监听器，当监听到diamond的数据发生改变的时候，就会通知监听器接收改变后的数据。所以应用程序需要做的就是设置diamond的监听器。
Diamond.java
static public void addListener(String dataId, String group, ManagerListener listener)
{
defaultEnv.addListeners(dataId, group, Arrays.asList(listener));
}
在使用这个方法时，会首先执行静态代码块的内容
static {
try {
initLog();
checkSnapshotValidity();
}
catch (Exception e)
{
e.printStackTrace();
throw new RuntimeException(e);
}
}
static private void checkSnapshotValidity()
{
List<String> localServerlist = LocalConfigInfoProcessor.readServerlist(defaultEnv);
List<String> apacheServerlist = defaultEnv.getServerUrls();
log.info("[apache-urls] " + apacheServerlist);
log.info("[cache-urls] " + localServerlist);
boolean isNotChange = apacheServerlist.equals(localServerlist); if (isNotChange) {
log.info(LogConstants.PREFFIX + "environment ok.");
} else {
log.warn(LogConstants.PREFFIX + "environment changed. clear cache.");
LocalConfigInfoProcessor.cleanAllSnapshot();
LocalConfigInfoProcessor.saveServerlist(defaultEnv, apacheServerlist);
}
}
在这里又静态导入了defaultEnv，所以导致了defaultEnv的初始化。
static public final DiamondEnv defaultEnv = new DiamondEnv(new ServerListManager());
DiamondEnv 的初始化需要ServerListManager的实例。
protected DiamondEnv(ServerListManager serverListMgr) {
{
initServerManager(serverListMgr); //初始化ServerListManager
cacheMap = new AtomicReference<Map<String, CacheData>>(new HashMap<String, CacheData>()); //初始化放CacheData的容器
worker = new ClientWorker(this); //初始化ClientWorker
}
看一下如何初始化的ServerListManager
public void initServerManager(ServerListManager _serverMgr)
{
_serverMgr.setEnv(this); //为ServerListManager 设置env
serverMgr = _serverMgr; //为DiamondEnv设置ServerListManager
serverMgr.start(); //ServerListManager 启动
agent = new ServerHttpAgent(serverMgr); //创建一个http的Client，用于同diamond server进行http的通信。
}
关键是serverMgr.start();
public synchronized void start()
{
if (isStarted || isFixed) { return; }
GetServerListTask getServersTask = new GetServerListTask(DiamondConfigure.singleton.getDiamondHttpUrl());
while (serverUrls.isEmpty()) {
getServersTask.run();
try {
Thread.sleep(1000L); }
catch (Exception e) {}
}
TimerService.scheduleWithFixedDelay(getServersTask, 0L, 30L,TimeUnit.SECONDS);
isStarted = true;
}
这里有一个定时的线程池每30s执行这个任务，这个任务就是去diamond服务器取服务器的ip列表。我们发现有一个DiamondConfigure.singleton.getDiamondHttpUrl()，所以我们需要在设置监听前要设置这个diamond的http url。这个很重要，要不然client起不来。
根据服务器上获取的ip列表与本地的进行比较，不一致的话
LocalConfigInfoProcessor.saveServerlist(name, serverUrls);
更新本地的ip列表，否则直接return。这是initServerManager所做的事情。我们再来看一下
worker = new ClientWorker(this);
这里面做的事情就比较多了。
ClientWorker(final DiamondEnv env) {
this.env = env; //ClientWorker中的env进行赋值
executor = Executors.newScheduledThreadPool(1, new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {
Thread t = new Thread(r);
t.setName("com.taobao.diamond.client.Worker."+ env.serverMgr.name);
t.setDaemon(true);
return t;
}
});
executor.scheduleWithFixedDelay(new Runnable() {
public void run() {
try {
checkLocalConfigInfo();
checkServerConfigInfo();
} catch (Throwable e) {
log.error("[sub-error-rotate] rotate check error", e);
}
}
}, 1L, 1L, TimeUnit.MILLISECONDS);
}
创建了一个定时的线程池，每隔1毫秒，进行checkLocalConfigInfo和checkServerConfigInfo，checkLocalConfigInfo是本地容灾文件的处理，
我们着重看一下checkServerConfigInfo。
static public void checkServerConfigInfo(DiamondEnv env) {
for (String groupKey : checkUpdateDataIds(env)) {
String dataId = GroupKey.parseKey(groupKey)[0];
String group = GroupKey.parseKey(groupKey)[1];
try {
String content = getServerConfig(env, dataId, group, 3000L);
CacheData cache = env.getCache(dataId, group);
cache.setContent(content);
log.info("[data-received] dataId=" + dataId + ", group=" + group + ", md5="
+ cache.getMd5() + ", content=" + ContentUtils.truncateContent(content));
} catch (IOException ioe) {
log.warn(ioe.toString(), ioe);
}
}
checkListenerMd5(env);
}
checkUpdateDataIds 是将本地缓存的CacheData中的group,dataId和md5作为字符串拼接起来，作为参数，post到服务器上进行处理，根据group和dataId，查询到md5与传递过来的MD5进行比对，如果不一致，将不一致的group和dataId回传回来。
/**
* 从DiamondServer获取值变化了的DataID列表。返回的对象里只有dataId和group是有效的。 保证不返回NULL。
*/
static List<String> checkUpdateDataIds(DiamondEnv env) {
/*
if (MockServer.isTestMode()) {
List<String> updateList = new ArrayList<String>();
//与DiamondEnv的模拟数据源比较，得出数据变化列表
for(CacheData cacheData : env.getAllCacheDataSnapshot()){
CacheData mockServerData = env.getMockCache(cacheData.dataId, cacheData.group);
if(mockServerData == null || !mockServerData.getMd5().equals(cacheData.getMd5()))
updateList.add(GroupKey.getKey(cacheData.dataId, cacheData.group));
}
return updateList;
} */
if (MockServer.isTestMode()) {
// 避免 test mode cpu% 过高
try {
Thread.sleep(3000l);
} catch (InterruptedException e) {}
List<String> updateList = new ArrayList<String>();
for(CacheData cacheData : env.getAllCacheDataSnapshot()){
if(!CacheData.getMd5String(MockServer.getConfigInfo(cacheData.dataId, cacheData.group, env))
.equals(cacheData.getMd5())) {
updateList.add(GroupKey.getKey(cacheData.dataId, cacheData.group));
}
}
return updateList;
}
String probeUpdateString = getProbeUpdateString(env);
List<String> params = Arrays.asList(Constants.PROBE_MODIFY_REQUEST, probeUpdateString);
long timeout = TimeUnit.SECONDS.toMillis(30L);
List<String> headers = Arrays.asList("longPullingTimeout", "" + timeout);
if (StringUtils.isBlank(probeUpdateString)) {
return Collections.emptyList();
}
try {
HttpResult result = env.agent.httpPost("/config.co", headers, params, Constants.ENCODE,
timeout);
if (HttpURLConnection.HTTP_OK == result.code) {
return parseUpdateDataIdResponse(result.content);
} else {
log.warn("[check-update] get changed dataId error, HTTP State: " + result.code);
}
} catch (IOException e) {
log.warn("[check-update] get changed dataId exception, " + e.toString());
}
return Collections.emptyList();
}
那么回传回来的group和dataId说明是有变化的，随后再请求服务器中的数据（最新的数据），放入本地的snapshot文件再放入放入CacheData
String content = getServerConfig(env, dataId, group, 3000L);
CacheData cache = env.getCache(dataId, group);
cache.setContent(content);
/**
* 对于404响应码，返回NULL.
*
* @throws IOException
*/
static String getServerConfig(DiamondEnv env, String dataId, String group, long readTimeout)
throws IOException {
if (StringUtils.isBlank(group)) {
group = Constants.DEFAULT_GROUP;
}
if (MockServer.isTestMode()) {
return MockServer.getConfigInfo(dataId, group, env);
}
HttpResult result = null;
try {
List<String> params = Arrays.asList("dataId", dataId, "group", group);
result = env.agent.httpGet("/config.co", null, params, Constants.ENCODE, readTimeout);
} catch (IOException e) {
log.warn("[sub-server] get server config exception, dataId=" + dataId + ", group="
+ group + ", " + e.toString());
throw e;
}
switch (result.code) {
case HttpURLConnection.HTTP_OK:
// if (env == defaultEnv) {
LocalConfigInfoProcessor.saveSnapshot(env, dataId, group, result.content);
// }
return result.content;
case HttpURLConnection.HTTP_NOT_FOUND:
// if (env == defaultEnv) {
LocalConfigInfoProcessor.saveSnapshot(env, dataId, group, null);
// }
return null;
case HttpURLConnection.HTTP_CONFLICT: {
log.warn("[sub-server-error] data being modified");
throw new IOException("data being modified");
}
default: {
log.warn("[sub-server-error] error code " + result.code);
throw new IOException("http code: " + result.code);
}
}
}
CacheData处理完了，在处理监听器。
static void checkListenerMd5(DiamondEnv env) {
for (CacheData cacheData : env.getAllCacheDataSnapshot()) {
cacheData.checkListenerMd5();
}
}
void checkListenerMd5() {
for (ManagerListenerWrap wrap : listeners) {
if (!md5.equals(wrap.lastCallMd5)) {
safeNotifyListener(dataId, group, content, md5, wrap);
}
}
static void safeNotifyListener(final String dataId, final String group, final String content,
final String md5, ManagerListenerWrap listenerWrap) {
final ManagerListener listener = listenerWrap.listener;
listenerWrap.lastCallMd5 = md5;
Runnable job = new Runnable() {
public void run() {
try {
listener.receiveConfigInfo(content);
log.info("[notify-ok] " + dataId + ", " + group + ", md5=" + md5
+ ", listener=" + listener);
} catch (Throwable t) {
log.error("[notify-error] " + dataId + ", " + group + ", md5=" + md5
+ ", listener=" + listener.toString(), t);
}
}
};
try {
if (null != listener.getExecutor()) {
listener.getExecutor().execute(job);
} else {
job.run();
}
} catch (Throwable t) {
log.error("[notify-error] " + dataId + ", " + group + ", md5=" + md5 + ", listener="
+ listener.toString(), t);
}
}
每次调用safeNotifyListener时都会开启一个线程（有线程池用线程池），将改变后的内容传递给监听器进行处理。
网友评论