1 server
1.1 data initialization
After initializing PersistService, the data source will be initialized:
@PostConstruct
public void initDataSource() throws Exception {
// Read JDBC Properties configuration, load data source
Properties props = ResourceUtils.getResourceAsProperties("jdbc.properties");
BasicDataSource ds = new BasicDataSource();
ds.setDriverClassName(JDBC_DRIVER_NAME);
ds.setUrl(ensurePropValueNotNull(props.getProperty("db.url")));
ds.setUsername(ensurePropValueNotNull(props.getProperty("db.user")));
ds.setPassword(ensurePropValueNotNull(props.getProperty("db.password")));
ds.setInitialSize(Integer.parseInt(ensurePropValueNotNull(props.getProperty("db.initialSize"))));
ds.setMaxActive(Integer.parseInt(ensurePropValueNotNull(props.getProperty("db.maxActive"))));
ds.setMaxIdle(Integer.parseInt(ensurePropValueNotNull(props.getProperty("db.maxIdle"))));
ds.setMaxWait(Long.parseLong(ensurePropValueNotNull(props.getProperty("db.maxWait"))));
ds.setPoolPreparedStatements(Boolean.parseBoolean(ensurePropValueNotNull(props
.getProperty("db.poolPreparedStatements"))));
this.jt = new JdbcTemplate();
this.jt.setDataSource(ds);
// Set the maximum number of records to prevent memory expansion
this.jt.setMaxRows(MAX_ROWS);
// Set JDBC execution timeout
this.jt.setQueryTimeout(QUERY_TIMEOUT);
}In TimerTaskService, a scheduled task will be initialized to query the latest data from the database every 600 seconds. After querying the data, first update the cache and then write to the disk:
@PostConstruct
public void init() {
this.scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {
public Thread newThread(Runnable r) {
Thread t = new Thread(r);
t.setName(THREAD_NAME);
t.setDaemon(true);
return t;
}
});
DumpConfigInfoTask dumpTask = new DumpConfigInfoTask(this);
dumpTask.run();
this.scheduledExecutorService.scheduleWithFixedDelay(dumpTask, SystemConfig.getDumpConfigInterval(),
SystemConfig.getDumpConfigInterval(), TimeUnit.SECONDS);
}In NotifyService, the configuration file will be read and the server node will be loaded:
@PostConstruct
public void loadNodes() {
InputStream in = null;
try {
in = ResourceUtils.getResourceAsStream("node.properties");
nodeProperties.load(in);
}
catch (IOException e) {
log.error("Failed to load node profile");
}
finally {
try {
if (in != null)
in.close();
}
catch (IOException e) {
log.error("close node.properties fail", e);
}
}
log.info("Node list:" + nodeProperties);
}1.2 update configuration
The login process is omitted, and the submitted data enters the postConfig method of AdminController class.
First check the parameters, and then enter the specific processing operation.
The source code is as follows:
@RequestMapping(params = "method=postConfig", method = RequestMethod.POST)
public String postConfig(HttpServletRequest request, HttpServletResponse response,
@RequestParam("dataId") String dataId, @RequestParam("group") String group,
@RequestParam("content") String content, ModelMap modelMap) {
response.setCharacterEncoding("GBK");
boolean checkSuccess = true;
String errorMessage = "Parameter error";
if (StringUtils.isBlank(dataId) || DiamondUtils.hasInvalidChar(dataId.trim())) {
checkSuccess = false;
errorMessage = "invalid DataId";
}
if (StringUtils.isBlank(group) || DiamondUtils.hasInvalidChar(group.trim())) {
checkSuccess = false;
errorMessage = "Invalid grouping";
}
if (StringUtils.isBlank(content)) {
checkSuccess = false;
errorMessage = "Invalid content";
}
if (!checkSuccess) {
modelMap.addAttribute("message", errorMessage);
return "/admin/config/new";
}
dataId = dataId.trim();
group = group.trim();
this.configService.addConfigInfo(dataId, group, content);
modelMap.addAttribute("message", "Submitted successfully!");
return listConfig(request, response, dataId, group, 1, 20, modelMap);
}Enter the addConfigInfo method of the ConfigService class.
First save the data to the database, then update the cache, and then save to the local disk.
Finally, you need to notify other service nodes to synchronize changes.
The source code is as follows:
public void addConfigInfo(String dataId, String group, String content) {
checkParameter(dataId, group, content);
ConfigInfo configInfo = new ConfigInfo(dataId, group, content);
// Save order: database first, then disk
try {
persistService.addConfigInfo(configInfo);
// Remember to update the cache
this.contentMD5Cache.put(generateMD5CacheKey(dataId, group), configInfo.getMd5());
diskService.saveToDisk(configInfo);
// Notify other nodes
this.notifyOtherNodes(dataId, group);
}
catch (Exception e) {
log.error("preservation ConfigInfo fail", e);
throw new ConfigServiceException(e);
}
}1.3 node notification
In the ConfigService class, call the notifyConfigInfoChange method of the Service class.
After updating the configuration, you also need to notify other nodes to resynchronize the configuration from the database.
The source code is as follows:
private void notifyOtherNodes(String dataId, String group) {
this.notifyService.notifyConfigInfoChange(dataId, group);
}Enter the notifyconfiginfo change method of NotifyService class.
Call the interface of the notification node through the URL.
The source code is as follows:
public void notifyConfigInfoChange(String dataId, String group) {
Enumeration<?> enu = nodeProperties.propertyNames();
while (enu.hasMoreElements()) {
String address = (String) enu.nextElement();
if (address.contains(SystemConfig.LOCAL_IP)) {
continue;
}
String urlString = generateNotifyConfigInfoPath(dataId, group, address);
final String result = invokeURL(urlString);
log.info("Notification node" + address + "Packet information change:" + result);
}
}Enter the notifyConfigInfo method of NotifyController class.
Execute synchronization logic.
The source code is as follows:
@RequestMapping(method = RequestMethod.GET, params = "method=notifyConfigInfo")
public String notifyConfigInfo(@RequestParam("dataId") String dataId, @RequestParam("group") String group) {
dataId = dataId.trim();
group = group.trim();
this.configService.loadConfigInfoToDisk(dataId, group);
return "200";
}Enter the loadconfigininfodisk method of ConfigService class.
Update the cache first and then save to local disk.
The source code is as follows:
public void loadConfigInfoToDisk(String dataId, String group) {
try {
ConfigInfo configInfo = this.persistService.findConfigInfo(dataId, group);
if (configInfo != null) {
this.contentMD5Cache.put(generateMD5CacheKey(dataId, group), configInfo.getMd5());
this.diskService.saveToDisk(configInfo);
}
else {
// Delete file
this.contentMD5Cache.remove(generateMD5CacheKey(dataId, group));
this.diskService.removeConfigInfo(dataId, group);
}
}
catch (Exception e) {
log.error("preservation ConfigInfo To disk failed", e);
throw new ConfigServiceException(e);
}
}2 client
2.1 DiamondManager
Create DiamondManager through the DefaultDiamondManager constructor.
Get the configuration content through the getavailableconfigureinformation method of DiamondManager.
The source code is as follows:
@Test
void testClient() {
DiamondManager diamondManager = new DefaultDiamondManager("cnpark", "cnpark-station", new ManagerListener() {
public void receiveConfigInfo(String configInfo) {
System.out.println("receiveConfigInfo" + configInfo);
}
public Executor getExecutor() {
return null;
}
});
diamondManager.getDiamondConfigure().setPort(8080);
String availableConfigureInfomation = diamondManager.getAvailableConfigureInfomation(5000);
System.out.println("content" + availableConfigureInfomation);
}Construct the DefaultDiamondManager object.
Create and start DiamondSubscriber.
The source code is as follows:
public DefaultDiamondManager(String group, String dataId, ManagerListener managerListener) {
this.dataId = dataId;
this.group = group;
diamondSubscriber = DiamondClientFactory.getSingletonDiamondSubscriber();
this.managerListeners.add(managerListener);
((DefaultSubscriberListener) diamondSubscriber.getSubscriberListener()).addManagerListeners(this.dataId,
this.group, this.managerListeners);
diamondSubscriber.addDataId(this.dataId, this.group);
diamondSubscriber.start();
}2.2 ManagerListener
ManagerListener is an interface and its methods need to be implemented and rewritten.
The receiveConfigInfo method is used to receive configuration information. When the configuration information is changed, you can update the configuration information.
The source code is as follows:
public interface ManagerListener {
public Executor getExecutor();
public void receiveConfigInfo(final String configInfo);
}2.3 DiamondSubscriber
Create a DiamondSubscriber through DiamondClientFactory, which actually creates a DefaultDiamondSubscriber.
DiamondSubscriber is used to subscribe to persistent text configuration information.
The source code is as follows:
private static DiamondSubscriber diamondSubscriber = new DefaultDiamondSubscriber(new DefaultSubscriberListener());
public static DiamondSubscriber getSingletonDiamondSubscriber() {
return diamondSubscriber;
}Create DefaultDiamondSubscriber.
Pass in SubscriberListener and create DiamondConfigure.
The source code is as follows:
public DefaultDiamondSubscriber(SubscriberListener subscriberListener) {
this.subscriberListener = subscriberListener;
this.diamondConfigure = new DiamondConfigure();
}2.4 SubscriberListener
When you create a SubscriberListener, you actually create a DefaultSubscriberListener.
The source code is as follows:
public Executor getExecutor(); public void receiveConfigInfo(final ConfigureInfomation configureInfomation);
DefaultSubscriberListener maintains a Map set of ManagerListener. Key is the combination of dataId and group, and Value is the List set of ManagerListener.
The receiveConfigInfo method is rewritten. When the configuration is changed, the ManagerListener will be traversed and its receiveConfigInfo method will be executed.
The source code is as follows:
public void receiveConfigInfo(final ConfigureInfomation configureInfomation) {
String dataId = configureInfomation.getDataId();
String group = configureInfomation.getGroup();
if (null == dataId) {
dataLog.error("[receiveConfigInfo] dataId is null");
return;
}
String key = makeKey(dataId, group);
CopyOnWriteArrayList<ManagerListener> listeners = allListeners.get(key);
if (listeners == null || listeners.isEmpty()) {
dataLog.warn("[notify-listener] no listener for dataId=" + dataId + ", group=" + group);
return;
}
for (ManagerListener listener : listeners) {
try {
notifyListener(configureInfomation, listener);
}
catch (Throwable t) {
dataLog.error("call listener error, dataId=" + dataId + ", group=" + group, t);
}
}
}
private void notifyListener(final ConfigureInfomation configureInfomation, final ManagerListener listener) {
if (listener == null) {
return;
}
final String dataId = configureInfomation.getDataId();
final String group = configureInfomation.getGroup();
final String content = configureInfomation.getConfigureInfomation();
dataLog.info("[notify-listener] call listener " + listener.getClass().getName() + ", for " + dataId + ", "
+ group + ", " + content);
Runnable job = new Runnable() {
public void run() {
try {
listener.receiveConfigInfo(content);
}
catch (Throwable t) {
dataLog.error(t.getMessage(), t);
}
}
};
if (null != listener.getExecutor()) {
listener.getExecutor().execute(job);
}
else {
job.run();
}
}2.5 DiamondConfigure
Create DiamondConfigure.
DiamondConfigure encapsulates some configuration information of the client and creates a configuration directory on the disk.
The source code is as follows:
public DiamondConfigure() {
filePath = System.getProperty("user.home") + "/diamond";
File dir = new File(filePath);
dir.mkdirs();
if (!dir.exists()) {
throw new RuntimeException("establish diamond Directory failed:" + filePath);
}
}2.6 starting DiamondSubscriber
Start LocalConfigInfoProcessor and ServerAddressProcessor successively in the method, and finally poll to obtain the configuration.
The source code is as follows:
public synchronized void start() {
if (isRun) {
return;
}
if (null == scheduledExecutor || scheduledExecutor.isTerminated()) {
scheduledExecutor = Executors.newSingleThreadScheduledExecutor();
}
localConfigInfoProcessor.start(this.diamondConfigure.getFilePath() + "/" + DATA_DIR);
serverAddressProcessor = new ServerAddressProcessor(this.diamondConfigure, this.scheduledExecutor);
serverAddressProcessor.start();
this.snapshotConfigInfoProcessor =
new SnapshotConfigInfoProcessor(this.diamondConfigure.getFilePath() + "/" + SNAPSHOT_DIR);
// Set domainNamePos value
randomDomainNamePos();
initHttpClient();
// Initialization complete
isRun = true;
if (log.isInfoEnabled()) {
log.info("The domain names currently used are:" + this.diamondConfigure.getDomainNameList());
}
if (MockServer.isTestMode()) {
bFirstCheck = false;
}
else {
// Polling interval setting
this.diamondConfigure.setPollingIntervalTime(Constants.POLLING_INTERVAL_TIME);
}
// polling
rotateCheckConfigInfo();
addShutdownHook();
}2.7 start LocalConfigInfoProcessor
Create the data directory through the start method of LocalConfigInfoProcessor.
The source code is as follows:
public synchronized void start(String rootPath) {
if (this.isRun) {
return;
}
this.rootPath = rootPath;
this.isRun = true;
if (this.singleExecutor == null || singleExecutor.isTerminated()) {
singleExecutor = Executors.newSingleThreadScheduledExecutor();
}
initDataDir(rootPath);
startCheckLocalDir(rootPath);
}2.8 start ServerAddressProcessor
If it is local priority, get the ServerAddress through the local file. The default is non local priority.
If non local priority is given, obtain the ServerAddress synchronously for the first time, and then obtain the ServerAddress periodically and asynchronously. This step is performed by default.
The source code is as follows:
public synchronized void start() {
if (isRun) {
return;
}
isRun = true;
initHttpClient();
if (this.diamondConfigure.isLocalFirst()) {
acquireServerAddressFromLocal();
}
else {
synAcquireServerAddress();
asynAcquireServerAddress();
}
}2.8.1 get ServerAddress synchronously
Call the acquireserveraddresonce method, first from the online server, then from the daily server, and finally from the local file.
If successful from the server, update the local file.
The source code is as follows:
protected void synAcquireServerAddress() {
if (!isRun) {
throw new RuntimeException("ServerAddressProcessor Not running, unable to get the server address list synchronously");
}
if (MockServer.isTestMode()) {
diamondConfigure.addDomainName("Test mode, no real server used");
return;
}
int acquireCount = 0;
if (diamondConfigure.getDomainNameList().size() == 0) {
if (!acquireServerAddressOnce(acquireCount)) {
acquireCount++;
if (acquireServerAddressOnce(acquireCount)) {
// Save to local file
storeServerAddressesToLocal();
log.info("When obtaining the server list synchronously, report to the daily ConfigServer The server got the list of servers");
}
else {
log.info("Get from local Diamond Address list");
reloadServerAddresses();
if (diamondConfigure.getDomainNameList().size() == 0)
throw new RuntimeException("There is currently no list of servers available");
}
}
else {
log.info("When synchronously obtaining the server list, the online ConfigServer The server got the list of servers");
// Save to local file
storeServerAddressesToLocal();
}
}
}2.8.2 get ServiceAddress periodically and asynchronously
Call the acquireserveraddresonce method to obtain it from the online server first, and then from the daily server.
If successful from the server, update the local file.
Update every 300 seconds.
The source code is as follows:
protected void asynAcquireServerAddress() {
if (MockServer.isTestMode()) {
return;
}
this.scheduledExecutor.schedule(new Runnable() {
public void run() {
if (!isRun) {
log.warn("ServerAddressProcessor Not in running state, unable to get the server address list asynchronously");
return;
}
int acquireCount = 0;
if (!acquireServerAddressOnce(acquireCount)) {
acquireCount++;
if (acquireServerAddressOnce(acquireCount)) {
// Save to local file
storeServerAddressesToLocal();
}
}
else {
// Save to local file
storeServerAddressesToLocal();
}
asynAcquireServerAddress();
}
}, asynAcquireIntervalInSec, TimeUnit.SECONDS);
}2.8.3 get ServiceAddress from server
Get the server address and port, as well as the file name, and get the ServiceAddress saved in the file.
The source code is as follows:
private boolean acquireServerAddressOnce(int acquireCount) {
HostConfiguration hostConfiguration = configHttpClient.getHostConfiguration();
String configServerAddress;
int port;
if (null != diamondConfigure.getConfigServerAddress()) {
configServerAddress = diamondConfigure.getConfigServerAddress();
port = diamondConfigure.getConfigServerPort();
}
else {
if (acquireCount == 0) {
configServerAddress = Constants.DEFAULT_DOMAINNAME;
port = Constants.DEFAULT_PORT;
}
else {
configServerAddress = Constants.DAILY_DOMAINNAME;
port = Constants.DEFAULT_PORT;
}
}
hostConfiguration.setHost(configServerAddress, port);
String serverAddressUrl = Constants.CONFIG_HTTP_URI_FILE;
HttpMethod httpMethod = new GetMethod(serverAddressUrl);
// Set parameters of HttpMethod
HttpMethodParams params = new HttpMethodParams();
params.setSoTimeout(diamondConfigure.getOnceTimeout());
// ///
httpMethod.setParams(params);
try {
if (SC_OK == configHttpClient.executeMethod(httpMethod)) {
InputStreamReader reader = new InputStreamReader(httpMethod.getResponseBodyAsStream());
BufferedReader bufferedReader = new BufferedReader(reader);
String address = null;
List<String> newDomainNameList = new LinkedList<String>();
while ((address = bufferedReader.readLine()) != null) {
address = address.trim();
if (StringUtils.isNotBlank(address)) {
newDomainNameList.add(address);
}
}
if (newDomainNameList.size() > 0) {
log.debug("Update the list of servers used");
this.diamondConfigure.setDomainNameList(newDomainNameList);
return true;
}
}
else {
log.warn("No new server list available");
}
}
catch (HttpException e) {
log.error(getErrorMessage(configServerAddress) + ", " + e);
}
catch (IOException e) {
log.error(getErrorMessage(configServerAddress) + ", " + e);
}
catch (Exception e) {
log.error(getErrorMessage(configServerAddress) + ", " + e);
}
finally {
httpMethod.releaseConnection();
}
return false;
}2.8.4 save to local file
If the ServiceAddress is obtained from the server, you also need to save the read address to the ServiceAddress file.
The source code is as follows:
void storeServerAddressesToLocal() {
List<String> domainNameList = new ArrayList<String>(diamondConfigure.getDomainNameList());
PrintWriter printWriter = null;
BufferedWriter bufferedWriter = null;
try {
File serverAddressFile =
new File(generateLocalFilePath(this.diamondConfigure.getFilePath(), "ServerAddress"));
if (!serverAddressFile.exists()) {
serverAddressFile.createNewFile();
}
printWriter = new PrintWriter(serverAddressFile);
bufferedWriter = new BufferedWriter(printWriter);
for (String serveraddress : domainNameList) {
bufferedWriter.write(serveraddress);
bufferedWriter.newLine();
}
bufferedWriter.flush();
}
catch (Exception e) {
log.error("Failed to store server address to local file", e);
}
finally {
if (bufferedWriter != null) {
try {
bufferedWriter.close();
}
catch (IOException e) {
// ignore
}
}
if (printWriter != null) {
printWriter.close();
}
}
}2.8.5 read ServerAddress from local file
If it is not obtained from the server, it can only be read from the file.
The source code is as follows:
void reloadServerAddresses() {
FileInputStream fis = null;
InputStreamReader reader = null;
BufferedReader bufferedReader = null;
try {
File serverAddressFile =
new File(generateLocalFilePath(this.diamondConfigure.getFilePath(), "ServerAddress"));
if (!serverAddressFile.exists()) {
return;
}
fis = new FileInputStream(serverAddressFile);
reader = new InputStreamReader(fis);
bufferedReader = new BufferedReader(reader);
String address = null;
while ((address = bufferedReader.readLine()) != null) {
address = address.trim();
if (StringUtils.isNotBlank(address)) {
diamondConfigure.getDomainNameList().add(address);
}
}
bufferedReader.close();
reader.close();
fis.close();
}
catch (Exception e) {
log.error("Failed to get server address from local file", e);
}
finally {
if (bufferedReader != null) {
try {
bufferedReader.close();
}
catch (Exception e) {
}
}
if (reader != null) {
try {
reader.close();
}
catch (Exception e) {
}
}
if (fis != null) {
try {
fis.close();
}
catch (Exception e) {
}
}
}
}2.9 polling acquisition configuration
There are three ways to detect whether the configuration information has changed:
1) check the local configuration file, compare the configuration in the cache with the configuration in the local configuration file, update the cache if there is a change, then call the popConfigInfo method.
2) check the server configuration file, compare the configuration in the cache with the configuration in the server configuration file, update the cache if there is a change, then call the popConfigInfo method.
3) check the local disk snapshot. If you do not read the configuration from the local configuration file nor read the configuration from the server configuration file, get the configuration in the local disk snapshot and update the cache, then call the popConfigInfo method.
After invoking the popConfigInfo method, you will push the subscription information of the configuration changes to the listener, and then call the saveSnapshot method to update the disk snapshot.
The source code is as follows:
private void rotateCheckConfigInfo() {
scheduledExecutor.schedule(new Runnable() {
public void run() {
if (!isRun) {
log.warn("DiamondSubscriber If it is not running, exit the query cycle");
return;
}
try {
checkLocalConfigInfo();
checkDiamondServerConfigInfo();
checkSnapshot();
}
catch (Exception e) {
e.printStackTrace();
log.error("Abnormal loop detection", e);
}
finally {
rotateCheckConfigInfo();
}
}
}, bFirstCheck ? 60 : diamondConfigure.getPollingIntervalTime(), TimeUnit.SECONDS);
bFirstCheck = false;
}2.10 get configuration
Get the configuration on the server through the getavailableconfigureinformation method of DiamondManager.
In fact, the getavailableconfigureinformation method of DefaultDiamondManager is called.
The source code is as follows:
public String getAvailableConfigureInfomation(long timeout) {
return diamondSubscriber.getAvailableConfigureInfomation(dataId, group, timeout);
}Continue calling the getavailableconfigureinformation method of DiamondSubscriber.
In fact, the getavailableconfigureinformation method of DefaultDiamondSubscriber is called.
The source code is as follows:
public String getAvailableConfigureInfomation(String dataId, String group, long timeout) {
// Try to get configuration information from local and network first
try {
String result = getConfigureInfomation(dataId, group, timeout);
if (result != null && result.length() > 0) {
return result;
}
}
catch (Throwable t) {
log.error(t.getMessage(), t);
}
// Test mode does not use local dump
if (MockServer.isTestMode()) {
return null;
}
return getSnapshotConfiginfomation(dataId, group);
}2.10.1 get configuration from local and server
Call the getconfigureinformation method.
Try to get the configuration from the local configuration file. If successful, update the cache and update the local disk snapshot.
Try to get the configuration from the server configuration file. If successful, update the cache and update the local disk snapshot.
The source code is as follows:
public String getConfigureInfomation(String dataId, String group, long timeout) {
// Synchronous interface flow control
// flowControl();
if (null == group) {
group = Constants.DEFAULT_GROUP;
}
CacheData cacheData = getCacheData(dataId, group);
// Local configuration is preferred
try {
String localConfig = localConfigInfoProcessor.getLocalConfigureInfomation(cacheData, true);
if (localConfig != null) {
cacheData.incrementFetchCountAndGet();
saveSnapshot(dataId, group, localConfig);
return localConfig;
}
}
catch (IOException e) {
log.error("Error getting local configuration file", e);
}
// Failed to get local configuration, get from network
String result = getConfigureInfomation(dataId, group, timeout, false);
if (result != null) {
saveSnapshot(dataId, group, result);
cacheData.incrementFetchCountAndGet();
}
return result;
}2.10.2 get configuration from local disk snapshot
The snapshot configuration method is called.
The source code is as follows:
private String getSnapshotConfiginfomation(String dataId, String group) {
if (group == null) {
group = Constants.DEFAULT_GROUP;
}
try {
CacheData cacheData = getCacheData(dataId, group);
String config = this.snapshotConfigInfoProcessor.getConfigInfomation(dataId, group);
if (config != null && cacheData != null) {
cacheData.incrementFetchCountAndGet();
}
return config;
}
catch (Exception e) {
log.error("obtain snapshot Error, dataId=" + dataId + ",group=" + group, e);
return null;
}
}