Zookeeper统一配置管理

た入场券 2022-08-13 17:45 248阅读 0赞

统一配置类  
import java.io.IOException;  
import java.util.ArrayList;  
import java.util.HashMap;  
import java.util.Iterator;  
import java.util.List;  
import java.util.Properties;  
import java.util.Set;  
import java.util.concurrent.CountDownLatch;  
import org.apache.log4j.Logger;  
import org.apache.zookeeper.CreateMode;  
import org.apache.zookeeper.KeeperException;  
import org.apache.zookeeper.WatchedEvent;  
import org.apache.zookeeper.Watcher;  
import org.apache.zookeeper.Watcher.Event.KeeperState;  
import org.apache.zookeeper.ZooKeeper;  
import org.apache.zookeeper.ZooDefs.Ids;  
import org.apache.zookeeper.ZooKeeper.States;  
import org.apache.zookeeper.data.ACL;  
import org.apache.zookeeper.data.Stat;  
import org.springframework.context.ApplicationContext;  
import org.springframework.context.support.ClassPathXmlApplicationContext;  
import com.ai.paas.PaasException;  
import com.ai.paas.util.StringUtil;  
  
  
/\*\*  
 \* 统一配置ZK管理类，实现初始化自动创建  
 \*  
 \*/  
public class ConfigurationCenter \{  
private static final Logger log = Logger  
.getLogger(ConfigurationCenter.class);  
private final String UNIX\_FILE\_SEPARATOR = "/";  
private CountDownLatch latch = new CountDownLatch(1);  
  
  
private ZooKeeper zk = null;  
  
  
private String centerAddr = null;  
private boolean createZKNode = false;  
  
  
private int timeOut = 2000;  
  
  
private String runMode = PROD\_MODE;// P:product mode; D:dev mode  
public static final String DEV\_MODE = "D";  
public static final String PROD\_MODE = "P";  
private List<String> configurationFiles = new ArrayList<String>();  
private Properties props = new Properties();  
  
  
private String PROP\_KEY\_ZK\_AUTH = "com.ai.auth.login.config";  
  
  
private HashMap<String, ArrayList<ConfigurationWatcher>> subsMap = null;  
  
  
public ConfigurationCenter(String centerAddr, int timeOut, String runMode,  
List<String> configurationFiles) \{  
this.centerAddr = centerAddr;  
this.timeOut = timeOut;  
this.runMode = runMode;  
if (configurationFiles != null) \{  
this.configurationFiles.addAll(configurationFiles);  
\}  
\}  
  
  
public ConfigurationCenter(String centerAddr, int timeOut, String runMode) \{  
this.centerAddr = centerAddr;  
this.timeOut = timeOut;  
this.runMode = runMode;  
\}  
  
  
public String getConfAndWatch(String confPath, ConfigurationWatcher warcher)  
throws PaasException \{  
ArrayList<ConfigurationWatcher> watcherList = subsMap.get(confPath);  
if (watcherList == null) \{  
watcherList = new ArrayList<ConfigurationWatcher>();  
subsMap.put(confPath, watcherList);  
\}  
watcherList.add(warcher);  
return this.getConf(confPath);  
\}  
  
  
public String getConf(String confPath) throws PaasException \{  
String conf = null;  
try \{  
if (DEV\_MODE.equals(this.getRunMode())) \{  
return props.getProperty(confPath);  
\} else \{  
if (null == zk || States.CONNECTED != zk.getState()) \{  
// 此时断掉了，需要重连  
zk = connectZookeeper(centerAddr, timeOut, runMode);  
\}  
conf = new String(zk.getData(confPath, true, null), "UTF-8");  
\}  
  
  
\} catch (Exception e) \{  
log.error("", e);  
throw new PaasException("9999",  
"failed to get configuration from configuration center", e);  
\}  
return conf;  
\}  
  
  
private synchronized ZooKeeper connectZookeeper(String address,  
int timeout, String runMode) throws Exception \{  
if (null != zk && States.CONNECTED == zk.getState())  
return zk;  
if (DEV\_MODE.equals(runMode)) \{  
zk = new ZooKeeper(centerAddr, timeout, new Watcher() \{  
public void process(WatchedEvent event) \{  
// 不做处理，节点变化也可能反映出来  
// 连接建立时, 打开latch, 唤醒wait在该latch上的线程  
if (event.getState() == KeeperState.SyncConnected) \{  
latch.countDown();  
\}  
\}  
\});  
// 增加认证信息  
Object auth = props.getProperty(PROP\_KEY\_ZK\_AUTH);  
if (auth != null) \{  
zk.addAuthInfo("digest", auth.toString().getBytes());  
\}  
return zk;  
\} else \{  
zk = new ZooKeeper(centerAddr, timeout, new Watcher() \{  
public void process(WatchedEvent event) \{  
if (log.isInfoEnabled()) \{  
log.info(event.toString());  
\}  
// 不做处理，节点变化也可能反映出来  
// 连接建立时, 打开latch, 唤醒wait在该latch上的线程  
if (event.getState() == KeeperState.SyncConnected) \{  
latch.countDown();  
\}  
if (Event.EventType.NodeDataChanged.equals(event.getType())  
&& subsMap.size() > 0) \{  
String path = event.getPath();  
ArrayList<ConfigurationWatcher> watcherList = subsMap  
.get(path);  
if (watcherList != null && watcherList.size() > 0) \{  
for (ConfigurationWatcher watcher : watcherList) \{  
try \{  
watcher.process(getConf(path));  
\} catch (PaasException e) \{  
e.printStackTrace();  
\}  
\}  
\}  
\}  
\}  
\});  
// 增加认证信息  
Object auth = props.getProperty(PROP\_KEY\_ZK\_AUTH);  
if (auth != null) \{  
zk.addAuthInfo("digest", auth.toString().getBytes());  
\}  
return zk;  
\}  
\}  
  
  
public void init() \{  
try \{  
for (String configurationFile : configurationFiles) \{  
props.load(this.getClass().getResourceAsStream(  
configurationFile));  
\}  
\} catch (IOException e) \{  
log.error("Error load proerpties file," + configurationFiles, e);  
\}  
// 增加watch,开发模式没有，生产有  
try \{  
zk = connectZookeeper(centerAddr, timeOut, runMode);  
\} catch (Exception e) \{  
log.error("Error connect to Zookeeper," + centerAddr, e);  
\}  
subsMap = new HashMap<String, ArrayList<ConfigurationWatcher>>();  
if (isCreateZKNode()) \{  
try \{  
latch.await();  
\} catch (InterruptedException e) \{  
e.printStackTrace();  
\}  
writeData();  
\}  
\}  
  
  
@SuppressWarnings("rawtypes")  
private void writeData() \{  
// 开始创建节点  
Set keyValue = props.keySet();  
for (Iterator it = keyValue.iterator(); it.hasNext();) \{  
String path = (String) it.next();  
String pathValue = (String) props.getProperty(path);  
// 开始创建  
try \{  
if (!path.equalsIgnoreCase(PROP\_KEY\_ZK\_AUTH)) \{  
setZKPathNode(zk, path, pathValue);  
\}  
\} catch (Exception e) \{  
log.error("Error create to set node data,key=" + path  
\+ ",value=" + pathValue, e);  
\}  
\}  
\}  
  
  
private void setZKPathNode(ZooKeeper zk, String path, String pathValue)  
throws Exception \{  
if (zk.exists(path, false) == null) \{  
createPathNode(zk, path.split(UNIX\_FILE\_SEPARATOR));  
\}  
// 设置值,匹配所有版本  
zk.setData(path, pathValue.getBytes("UTF-8"), -1);  
log.info("Set zk node data: node=" + path + ",value=" + pathValue);  
\}  
  
  
private void createPathNode(ZooKeeper zk, String\[\] pathParts)  
throws Exception \{  
StringBuilder path = new StringBuilder();  
for (int i = 0; i < pathParts.length; i++) \{  
if (!StringUtil.isBlank(pathParts\[i\])) \{  
path.append(UNIX\_FILE\_SEPARATOR).append(pathParts\[i\]);  
String pathString = path.toString();  
try \{  
if (zk.exists(pathString, false) == null) \{  
// 前面都是空  
zk.create(pathString, null, Ids.OPEN\_ACL\_UNSAFE,  
CreateMode.PERSISTENT);  
\}  
\} catch (KeeperException e) \{  
if (e.code() != KeeperException.Code.NODEEXISTS)  
throw e;  
\}  
\}  
\}  
\}  
  
  
@SuppressWarnings("rawtypes")  
public void removeWatcher(String confPath, Class warcherClazz)  
throws PaasException \{  
ArrayList<ConfigurationWatcher> watcherList = subsMap.get(confPath);  
try \{  
if (watcherList == null) \{  
zk.getData(confPath, false, null);  
\} else \{  
int size = watcherList.size();  
// ConfigurationWatcher watcher = null;  
for (int i = size - 1; i >= 0; i--) \{  
if (watcherList.get(i).getClass().equals(warcherClazz)) \{  
watcherList.remove(i);  
\}  
\}  
if (watcherList.size() == 0) \{  
zk.getData(confPath, false, null);  
\}  
\}  
\} catch (Exception e) \{  
log.error("", e);  
throw new PaasException("9999",  
"failed to get configuration from configuration center", e);  
\}  
  
  
\}  
  
  
@SuppressWarnings("rawtypes")  
public void addPathAuth(String auth, List<ACL> acls) throws Exception \{  
zk.addAuthInfo("digest", auth.toString().getBytes());  
Set keyValue = props.keySet();  
List<ACL> existACL = null;  
for (Iterator it = keyValue.iterator(); it.hasNext();) \{  
String path = (String) it.next();  
if (!path.equalsIgnoreCase(PROP\_KEY\_ZK\_AUTH)) \{  
existACL = null;  
try \{  
Stat stat = zk.exists(path, false);  
existACL = zk.getACL(path, stat);  
\} catch (Exception e) \{  
  
  
\}  
if (null != existACL) \{  
// 还需要去掉所有anyone的  
for (ACL acl : existACL) \{  
if (!acl.getId().getId().equalsIgnoreCase("anyone")) \{  
acls.add(acl);  
\}  
\}  
\}  
zk.setACL(path, acls, -1);  
\}  
\}  
\}  
  
  
public String getRunMode() \{  
return runMode;  
\}  
  
  
public List<String> getConfigurationFiles() \{  
return configurationFiles;  
\}  
  
  
public void setConfigurationFile(List<String> configurationFile) \{  
this.configurationFiles.addAll(configurationFiles);  
\}  
  
  
public void destory() \{  
if (null != zk) \{  
try \{  
log.info("Start to closing zk client," + zk);  
zk.close();  
log.info("ZK client closed," + zk);  
\} catch (InterruptedException e) \{  
log.error("Can not close zk client", e);  
\}  
\}  
\}  
  
  
public int getTimeOut() \{  
return timeOut;  
\}  
  
  
public void setTimeOut(int timeOut) \{  
this.timeOut = timeOut;  
\}  
  
  
public boolean isCreateZKNode() \{  
return createZKNode;  
\}  
  
  
public void setCreateZKNode(boolean createZKNode) \{  
this.createZKNode = createZKNode;  
\}  
  
  
public ZooKeeper getZk() \{  
return zk;  
\}  
  
  
public void setZk(ZooKeeper zk) \{  
this.zk = zk;  
\}  
  
  
@SuppressWarnings("resource")  
public static void main(String\[\] args) throws PaasException \{  
ApplicationContext ctx = new ClassPathXmlApplicationContext(  
new String\[\] \{ "paasContext.xml" \});  
ConfigurationCenter confCenter = (ConfigurationCenter) ctx  
.getBean("confCenter");  
log.error(confCenter.getConf("/com/ai/paas/session/conf"));  
// MongoLogWriter logWriter = (MongoLogWriter)ctx.getBean("logWriter");  
// while(true) \{  
// try \{  
// Thread.sleep(1000);  
// \} catch (InterruptedException e) \{  
// e.printStackTrace();  
// \}  
// \}  
\}

mongoDB日志写入工具类  
  
import java.util.Date;  
import java.util.Map;  
import net.sf.json.JSONObject;  
import org.apache.log4j.Logger;  
import org.springframework.context.ApplicationContext;  
import org.springframework.context.support.ClassPathXmlApplicationContext;  
import com.ai.paas.PaasException;  
import com.ai.paas.config.ConfigurationCenter;  
import com.ai.paas.config.ConfigurationWatcher;  
import com.ai.paas.file.impl.MongoDBClient;  
import com.ai.paas.log.ILogWriter;  
import com.ai.paas.util.JSONValidator;  
  
  
public class MongoLogWriter implements ConfigurationWatcher, ILogWriter \{  
private static final Logger log = Logger.getLogger(MongoLogWriter.class);  
  
  
private String confPath = "/com/ai/paas/logger/conf";  
  
  
private static final String LOG\_SERVER\_KEY = "logServer";  
private static final String LOG\_REPO\_KEY = "logRepo";  
private static final String LOG\_PATH\_KEY = "logPath";  
private static final String USERNAME = "userName";  
private static final String PASSWORD = "password";  
  
  
private String logServer = null;  
private String logRepo = null;  
private String logPath = null;  
private String userName = null;  
private String password = null;  
private MongoDBClient mongo = null;

//所有使用统一配置的工具类包含统一配置中心类的实例。  
private ConfigurationCenter confCenter = null;  
  
  
public MongoLogWriter() \{  
  
  
\}  
  
  
public void init() \{  
try \{  
process(confCenter.getConfAndWatch(confPath, this));  
  
  
\} catch (PaasException e) \{  
e.printStackTrace();  
\}  
\}  
  
  
/\*  
\* \{ logServer : \[\{ip:  
\* '133.0.43.195',port:'27017'\},\{ip:'133.0.43.196',port:'27017'\},\{ip:'133.0.43.196',port:'27027'\}\],  
\* logRepo : 'aipayLogDB', logPath : 'aipayLogCollection' \}  
\*   
\* @see com.ai.paas.client.ConfigurationWatcher\#process(java.lang.String)  
\*/  
public void process(String conf) \{  
if (log.isInfoEnabled()) \{  
log.info("new log configuration is received: " + conf);  
\}  
try \{  
JSONObject json = JSONObject.fromObject(conf);  
boolean changed = false;  
if (JSONValidator.isChanged(json, LOG\_SERVER\_KEY, logServer)) \{  
changed = true;  
logServer = json.getString(LOG\_SERVER\_KEY);  
\}  
if (JSONValidator.isChanged(json, LOG\_REPO\_KEY, logRepo)) \{  
changed = true;  
logRepo = json.getString(LOG\_REPO\_KEY);  
\}  
if (JSONValidator.isChanged(json, USERNAME, userName)) \{  
changed = true;  
userName = json.getString(USERNAME);  
\}  
if (JSONValidator.isChanged(json, PASSWORD, password)) \{  
changed = true;  
password = json.getString(PASSWORD);  
\}  
if (JSONValidator.isChanged(json, LOG\_PATH\_KEY, logPath)) \{  
// changed = true;  
logPath = json.getString(LOG\_PATH\_KEY);  
\}  
if (changed) \{  
if (logServer != null) \{  
mongo = new MongoDBClient(logServer, logRepo, userName,  
password);  
if (log.isInfoEnabled()) \{  
log.info("log server address is changed to "  
\+ logServer);  
\}  
\}  
\}  
\} catch (Exception e) \{  
log.error("", e);  
\}  
\}  
  
  
public void write(JSONObject logJson) \{  
mongo.insertJSON(logRepo, logPath, logJson);  
\}  
  
  
@SuppressWarnings("rawtypes")  
public void write(Map logMap) \{  
mongo.insert(logRepo, logPath, logMap);  
\}  
  
  
public void write(String log) \{  
mongo.insert(logRepo, logPath, log);  
\}  
  
  
public ConfigurationCenter getConfCenter() \{  
return confCenter;  
\}  
  
  
public void setConfCenter(ConfigurationCenter confCenter) \{  
this.confCenter = confCenter;  
\}  
  
  
public String getConfPath() \{  
return confPath;  
\}  
  
  
public void setConfPath(String confPath) \{  
this.confPath = confPath;  
\}  
  
  
@SuppressWarnings("resource")  
public static void main(String\[\] args) throws PaasException \{  
ApplicationContext ctx = new ClassPathXmlApplicationContext(  
new String\[\] \{ "paasContext.xml" \});  
ILogWriter logWriter = (MongoLogWriter) ctx.getBean("logWriter");  
for (int i = 0; i < 10; i++) \{  
System.out.println(i);  
String log = "\{level:'ERROR',server:'10.1.1.3',log:'test log "  
\+ new Date() + "'\}";  
logWriter.write(log);  
JSONObject json = JSONObject  
.fromObject("\{level:'ERROR是倒萨大',server:'10.1.1.4',log:'test log"  
\+ new Date() + "'\}");  
logWriter.write(json);  
try \{  
Thread.sleep(1000);  
\} catch (InterruptedException e) \{  
e.printStackTrace();  
\}  
\}  
  
  
while (true) \{  
try \{  
Thread.sleep(1000);  
\} catch (InterruptedException e) \{  
e.printStackTrace();  
\}  
\}  
\}  
\}

使用spring实例化加载配置中心以及工具类。

<?xml version="1.0" encoding="UTF-8"?>  
  
  
<beans xmlns="http://www.springframework.org/schema/beans"  
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:dubbo="http://code.alibabatech.com/schema/dubbo"  
xsi:schemaLocation="http://www.springframework.org/schema/beans  
        http://www.springframework.org/schema/beans/spring-beans.xsd  
        http://code.alibabatech.com/schema/dubbo  
        http://code.alibabatech.com/schema/dubbo/dubbo.xsd  
        ">  
<bean id="confCenter" class="com.ai.paas.config.ConfigurationCenter" init-method="init" destroy-method="destory">  
<constructor-arg value="127.0.0.1:2181" />  
<constructor-arg value="10000" />  
<constructor-arg value="P" />  
<constructor-arg>  
<list>  
<value>/paasConf.properties</value>  
<value>/saasConf.properties</value>  
</list>  
</constructor-arg>  
<property name="createZKNode" value="true" />  
</bean>  
  
  
<bean id="logWriter" class="com.ai.paas.log.impl.MongoLogWriter"  
init-method="init">  
<property name="confCenter" ref="confCenter" />  
<property name="confPath" value="/com/ai/paas/logger/conf" />  
</bean>  
  
  
  
</beans>

其中统一配置的所有工具类实例化所需的参数都存放在/paasConf.properties，其中值为json串

/com/ai/paas/logger/conf \{"logServer":\[\{"ip":"133.0.192.193","port":"27017"\}\],"logRepo":"admin","logPath":"woegoLogCollection","userName":"sa","password":"f938d82df85971d2"\}

关于zookeeper统一配置中心代码的分析：

1、核心在于zookeeper的监听的注册，在创建zookeeper时加入的监听主要实现两点：一则监听zookeeper的状态，如果zookeeper实现同步，则则释放线程锁，latch.countdown()；执行之后向zookeeper服务器写入配置信息的操作。二则，监听zookeeper的状态，如果zookeeper上的配置信息发生变化，则执行所有使用配置应用的注册的监听器的回调函数，即watcher，process(),将配置信息更新应用实例对象。

2、对于应用监听器的注册，在应用第一次获取配置信息的时候将自身作为监听器注册进去。即getConfAndWatch().

3、对于zookeeper的配置信息写入时的一些注意事项：写入路径的时候，是逐层写入，而不是一次性写入，createPathNode()方法实现该过程。