add NodeListener mechanism

This commit is contained in:
DigitalSonic
2020-01-22 16:48:16 +08:00
parent dd2b469414
commit 82adaad442
9 changed files with 822 additions and 0 deletions
@@ -0,0 +1,120 @@
/*
* Copyright 1999-2020 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.druid.pool.ha.node;
import com.alibaba.druid.pool.ha.PropertiesUtils;
import com.alibaba.druid.support.logging.Log;
import com.alibaba.druid.support.logging.LogFactory;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
/**
* A NodeList that monitors the change of a file.
*
* @author DigitalSonic
*/
public class FileNodeListener extends NodeListener {
private final static Log LOG = LogFactory.getLog(FileNodeListener.class);
private String file = null;
private String prefix = "";
private int intervalSeconds = 60;
private ScheduledExecutorService executor;
/**
* Start a Scheduler to check the specified file.
*
* @see #setIntervalSeconds(int)
* @see #update()
*/
@Override
public void init() {
super.init();
if (intervalSeconds <= 0) {
intervalSeconds = 60;
}
executor = Executors.newScheduledThreadPool(1);
executor.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
try {
update();
} catch (Exception e) {
LOG.error("Can NOT update the node list.", e);
}
}
}, 0, intervalSeconds, TimeUnit.SECONDS);
}
/**
* Load the properties file and diff with the stored Properties.
*
* @return A List of the modification
*/
@Override
public List<NodeEvent> refresh() {
Properties originalProperties = PropertiesUtils.loadProperties(file);
List<String> nameList = PropertiesUtils.loadNameList(originalProperties, prefix);
Properties properties = new Properties();
for (String n : nameList) {
String url = originalProperties.getProperty(n + ".url");
String username = originalProperties.getProperty(n + ".username");
String password = originalProperties.getProperty(n + ".password");
if (url == null || url.isEmpty()
|| username == null || username.isEmpty()
|| password == null || password.isEmpty()) {
continue;
}
properties.setProperty(n + ".url", url);
properties.setProperty(n + ".username", username);
properties.setProperty(n + ".password", password);
}
List<NodeEvent> events = NodeEvent.getEventListFromProperties(getProperties(), properties);
if (events != null && !events.isEmpty()) {
setProperties(properties);
}
return events;
}
public int getIntervalSeconds() {
return intervalSeconds;
}
public void setIntervalSeconds(int intervalSeconds) {
this.intervalSeconds = intervalSeconds;
}
public String getFile() {
return file;
}
public void setFile(String file) {
this.file = file;
}
public String getPrefix() {
return prefix;
}
public void setPrefix(String prefix) {
this.prefix = prefix;
}
}
@@ -0,0 +1,118 @@
/*
* Copyright 1999-2020 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.druid.pool.ha.node;
import com.alibaba.druid.pool.ha.PropertiesUtils;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
/**
* Define the detail of a node update event.
*
* @author DigitalSonic
*/
public class NodeEvent {
private NodeEventType type;
private String nodeName;
private String url;
private String username;
private String password;
/**
* Diff the given two Properties.
*
* @return A List of AddEvent and DelEvent
*/
public static List<NodeEvent> getEventListFromProperties(Properties previous, Properties next) {
List<String> prevNames = PropertiesUtils.loadNameList(previous, "");
List<String> nextNames = PropertiesUtils.loadNameList(next, "");
List<String> namesToAdd = new ArrayList<String>();
List<String> namesToDel = new ArrayList<String>();
for (String n : prevNames) {
if (n != null && !n.trim().isEmpty() && !nextNames.contains(n)) {
namesToDel.add(n);
}
}
for (String n : nextNames) {
if (n != null && !n.trim().isEmpty() && !prevNames.contains(n)) {
namesToAdd.add(n);
}
}
List<NodeEvent> list = new ArrayList<NodeEvent>();
list.addAll(generateEvents(next, namesToAdd, NodeEventType.ADD));
list.addAll(generateEvents(previous, namesToDel, NodeEventType.DELETE));
return list;
}
private static List<NodeEvent> generateEvents(Properties properties, List<String> names, NodeEventType type) {
List<NodeEvent> list = new ArrayList<NodeEvent>();
for (String n : names) {
NodeEvent event = new NodeEvent();
event.setType(type);
event.setNodeName(n);
event.setUrl(properties.getProperty(n + ".url"));
event.setUsername(properties.getProperty(n + ".username"));
event.setPassword(properties.getProperty(n + ".password"));
list.add(event);
}
return list;
}
public NodeEventType getType() {
return type;
}
public void setType(NodeEventType type) {
this.type = type;
}
public String getNodeName() {
return nodeName;
}
public void setNodeName(String nodeName) {
this.nodeName = nodeName;
}
public String getUrl() {
return url;
}
public void setUrl(String url) {
this.url = url;
}
public String getUsername() {
return username;
}
public void setUsername(String username) {
this.username = username;
}
public String getPassword() {
return password;
}
public void setPassword(String password) {
this.password = password;
}
}
@@ -0,0 +1,25 @@
/*
* Copyright 1999-2020 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.druid.pool.ha.node;
/**
* Enum that specifies the type of a NodeEvent.
*
* @author DigitalSonic
*/
public enum NodeEventType {
ADD, DELETE
}
@@ -0,0 +1,93 @@
/*
* Copyright 1999-2020 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.druid.pool.ha.node;
import java.util.Date;
import java.util.List;
import java.util.Observable;
import java.util.Observer;
import java.util.Properties;
/**
* This abstract class defines a listener to monitor the change of DataSource nodes.
*
* @author DigitalSonic
* @see Observable
*/
public abstract class NodeListener extends Observable {
private Properties properties = new Properties();
private Date lastUpdateTime = null;
private Observer observer = null;
/**
* The method implements the detail update logic.
*/
public abstract List<NodeEvent> refresh();
/**
* Add the given PoolUpdater as the Observer.
*
* @see #setObserver(Observer)
*/
public void init() {
if (observer == null) {
throw new RuntimeException("No Observer(such as PoolUpdater) specified.");
}
this.addObserver(observer);
}
/**
* Fire the refresh() method and notify the Observer.
*
* @see #refresh()
*/
public void update() {
List<NodeEvent> events = refresh();
if (events != null && !events.isEmpty()) {
this.lastUpdateTime = new Date();
NodeEvent[] arr = new NodeEvent[events.size()];
for (int i = 0; i < events.size(); i++) {
arr[i] = events.get(i);
}
this.setChanged();
this.notifyObservers(arr);
}
}
public Observer getObserver() {
return observer;
}
public void setObserver(Observer observer) {
this.observer = observer;
}
public Date getLastUpdateTime() {
return lastUpdateTime;
}
public void setLastUpdateTime(Date lastUpdateTime) {
this.lastUpdateTime = lastUpdateTime;
}
public Properties getProperties() {
return properties;
}
public void setProperties(Properties properties) {
this.properties = properties;
}
}
@@ -0,0 +1,201 @@
/*
* Copyright 1999-2020 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.druid.pool.ha.node;
import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.ha.DataSourceCreator;
import com.alibaba.druid.pool.ha.HighAvailableDataSource;
import com.alibaba.druid.support.logging.Log;
import com.alibaba.druid.support.logging.LogFactory;
import javax.sql.DataSource;
import java.util.HashSet;
import java.util.Map;
import java.util.Observable;
import java.util.Observer;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
/**
* Update the DataSource Connection Pool when notified.
*
* @author DigitalSonic
*/
public class PoolUpdater implements Observer {
private final static Log LOG = LogFactory.getLog(PoolUpdater.class);
private Set<String> nodesToDel = new CopyOnWriteArraySet<String>();
private HighAvailableDataSource highAvailableDataSource;
private Lock lock = new ReentrantLock();
private ScheduledExecutorService executor;
public PoolUpdater() {
executor = Executors.newScheduledThreadPool(1);
executor.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
try {
removeDataSources();
} catch (Exception e) {
LOG.error("Exception occurred while removing DataSources.", e);
}
}
}, 60, 60, TimeUnit.SECONDS);
}
public PoolUpdater(HighAvailableDataSource highAvailableDataSource) {
this();
setHighAvailableDataSource(highAvailableDataSource);
}
@Override
public void update(Observable o, Object arg) {
if (!(o instanceof NodeListener)) {
return;
}
if (arg == null || !(arg instanceof NodeEvent[])) {
return;
}
NodeEvent[] events = (NodeEvent[]) arg;
if (events.length <= 0) {
return;
}
try {
LOG.info("Waiting for Lock to start processing NodeEvents.");
lock.lock();
LOG.info("Start processing the NodeEvent[" + events.length + "].");
for (NodeEvent e : events) {
if (e.getType() == NodeEventType.ADD) {
addNode(e);
} else if (e.getType() == NodeEventType.DELETE) {
deleteNode(e);
}
}
} catch (Exception e) {
LOG.error("Exception occurred while updating Pool.", e);
} finally {
lock.unlock();
}
}
public void removeDataSources() {
if (nodesToDel == null || nodesToDel.isEmpty()) {
return;
}
try {
lock.lock();
Map<String, DataSource> map = highAvailableDataSource.getDataSourceMap();
Set<String> copySet = new HashSet<String>(nodesToDel);
for (String nodeName : copySet) {
LOG.info("Start removing Node " + nodeName + ".");
if (!map.containsKey(nodeName)) {
LOG.info("Node " + nodeName + " is NOT existed in the map.");
cancelBlacklistNode(nodeName);
continue;
}
DataSource ds = map.get(nodeName);
if (ds instanceof DruidDataSource) {
DruidDataSource dds = (DruidDataSource) ds;
int activeCount = dds.getActiveCount(); // CAUTION, activeCount MAYBE changed!
if (activeCount > 0) {
LOG.warn("Node " + nodeName + " is still running [activeCount=" + activeCount
+ "], try next time.");
continue;
} else {
LOG.info("Close Node " + nodeName + " and remove it.");
try {
dds.close();
} catch (Exception e) {
LOG.error("Exception occurred while closing Node " + nodeName
+ ", just remove it.", e);
}
}
}
map.remove(nodeName); // Remove the node directly if it is NOT a DruidDataSource.
cancelBlacklistNode(nodeName);
}
} catch (Exception e) {
LOG.error("Exception occurred while removing DataSources.", e);
} finally {
lock.unlock();
}
}
protected void addNode(NodeEvent event) {
String nodeName = event.getNodeName();
String url = event.getUrl();
String username = event.getUsername();
String password = event.getPassword();
Map<String, DataSource> map = highAvailableDataSource.getDataSourceMap();
if (nodeName == null || nodeName.isEmpty()) {
return;
}
LOG.info("Adding Node " + nodeName + "[url: " + url + ", username: " + username + "].");
if (map.containsKey(nodeName)) {
cancelBlacklistNode(nodeName);
return;
}
try {
DruidDataSource dataSource = DataSourceCreator.create(nodeName, url, username,
password, this.highAvailableDataSource);
map.put(nodeName, dataSource);
} catch (Exception e) {
LOG.error("Can NOT create DataSource " + nodeName, e);
}
}
protected void deleteNode(NodeEvent event) {
String nodeName = event.getNodeName();
Map<String, DataSource> map = highAvailableDataSource.getDataSourceMap();
if (nodeName == null || nodeName.isEmpty() || !map.containsKey(nodeName)) {
return;
}
blacklistNode(nodeName);
}
private void cancelBlacklistNode(String nodeName) {
LOG.info("Cancel the deletion of Node " + nodeName + ".");
nodesToDel.remove(nodeName);
highAvailableDataSource.removeBlackList(nodeName);
}
private void blacklistNode(String nodeName) {
LOG.info("Deleting Node " + nodeName + ", just add into blacklist.");
nodesToDel.add(nodeName);
highAvailableDataSource.addBlackList(nodeName);
}
public HighAvailableDataSource getHighAvailableDataSource() {
return highAvailableDataSource;
}
public void setHighAvailableDataSource(HighAvailableDataSource highAvailableDataSource) {
this.highAvailableDataSource = highAvailableDataSource;
}
public Set<String> getNodesToDel() {
return nodesToDel;
}
}
@@ -0,0 +1,71 @@
package com.alibaba.druid.pool.ha.node;
import org.junit.Test;
import sun.java2d.SurfaceDataProxy;
import java.util.List;
import java.util.Observable;
import java.util.Observer;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import static org.junit.Assert.*;
public class FileNodeListenerTest {
@Test
public void testUpdate() throws InterruptedException {
final CountDownLatch cdl = new CountDownLatch(1);
String file = "/com/alibaba/druid/pool/ha/ha-with-prefix-datasource.properties";
FileNodeListener listener = new FileNodeListener();
listener.setFile(file);
listener.setPrefix("prefix1");
listener.setObserver(new Observer() {
@Override
public void update(Observable o, Object arg) {
cdl.countDown();
assertTrue(o instanceof FileNodeListener);
assertTrue(arg instanceof NodeEvent[]);
NodeEvent[] events = (NodeEvent[]) arg;
assertEquals(1, events.length);
assertEquals(NodeEventType.ADD, events[0].getType());
assertEquals("prefix1.foo", events[0].getNodeName());
assertEquals("jdbc:derby:memory:foo1;create=true", events[0].getUrl());
}
});
listener.init();
listener.update();
assertTrue(cdl.await(100, TimeUnit.MILLISECONDS));
}
@Test
public void testRefresh_emptyPropertiesMatch() {
String file = "/com/alibaba/druid/pool/ha/ha-with-prefix-datasource.properties";
FileNodeListener listener = new FileNodeListener();
listener.setFile(file);
listener.setPrefix("prefix3");
List<NodeEvent> list = listener.refresh();
assertTrue(listener.getProperties().isEmpty());
assertTrue(list.isEmpty());
}
@Test
public void testRefresh() {
String file = "/com/alibaba/druid/pool/ha/ha-with-prefix-datasource.properties";
FileNodeListener listener = new FileNodeListener();
listener.setFile(file);
listener.setPrefix("prefix1");
List<NodeEvent> list = listener.refresh();
Properties properties = listener.getProperties();
assertEquals(3, properties.size());
assertEquals("jdbc:derby:memory:foo1;create=true", properties.getProperty("prefix1.foo.url"));
assertEquals(1, list.size());
NodeEvent event = list.get(0);
assertEquals(NodeEventType.ADD, event.getType());
assertEquals("prefix1.foo", event.getNodeName());
assertEquals("jdbc:derby:memory:foo1;create=true", event.getUrl());
}
}
@@ -0,0 +1,42 @@
package com.alibaba.druid.pool.ha.node;
import org.junit.Test;
import java.util.List;
import java.util.Properties;
import static org.junit.Assert.*;
public class NodeEventTest {
@Test
public void testGetEventListFromProperties_emptyProperties() {
Properties p1 = new Properties();
Properties p2 = new Properties();
List<NodeEvent> list = NodeEvent.getEventListFromProperties(p1, p2);
assertTrue(list.isEmpty());
p1.setProperty("foo.url", "foo_url");
p1.setProperty("foo.username", "foo_username");
p1.setProperty("foo.password", "foo_password");
list = NodeEvent.getEventListFromProperties(p1, p2);
assertEquals(1, list.size());
NodeEvent event = list.get(0);
assertEquals(NodeEventType.DELETE, event.getType());
assertEquals("foo", event.getNodeName());
assertEquals("foo_url", event.getUrl());
assertEquals("foo_username", event.getUsername());
assertEquals("foo_password", event.getPassword());
list = NodeEvent.getEventListFromProperties(p2, p1);
assertEquals(1, list.size());
event = list.get(0);
assertEquals(NodeEventType.ADD, event.getType());
assertEquals("foo", event.getNodeName());
assertEquals("foo_url", event.getUrl());
assertEquals("foo_username", event.getUsername());
assertEquals("foo_password", event.getPassword());
}
}
@@ -0,0 +1,148 @@
package com.alibaba.druid.pool.ha.node;
import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.ha.HighAvailableDataSource;
import com.alibaba.druid.pool.ha.MockDataSource;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import javax.sql.DataSource;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import static org.junit.Assert.*;
public class PoolUpdaterTest {
private HighAvailableDataSource haDataSource;
private PoolUpdater updater;
@Before
public void setUp() throws Exception {
haDataSource = new HighAvailableDataSource();
Map<String, DataSource> map = new HashMap<String, DataSource>();
haDataSource.setDataSourceMap(map);
updater = new PoolUpdater(haDataSource);
}
@After
public void tearDown() throws Exception {
haDataSource = null;
updater = null;
}
@Test
public void testUpdate() {
haDataSource.getDataSourceMap().put("foo", new MockDataSource("foo"));
NodeEvent event = new NodeEvent();
event.setNodeName("foo");
event.setType(NodeEventType.DELETE);
updater.update(new FileNodeListener(), new NodeEvent[] { event });
validateDeleteNode();
event = new NodeEvent();
event.setNodeName("foo");
event.setType(NodeEventType.ADD);
updater.update(new FileNodeListener(), new NodeEvent[] { event });
assertFalse(updater.getNodesToDel().contains("foo"));
assertFalse(haDataSource.isInBlackList("foo"));
}
@Test
public void testRemoveDataSources() {
String url = "jdbc:derby:memory:foo;create=true";
String name = "foo";
addNode(url, name);
DruidDataSource ds = (DruidDataSource) haDataSource.getDataSourceMap().get(name);
updater.getNodesToDel().add(name);
haDataSource.addBlackList(name);
updater.removeDataSources();
assertTrue(haDataSource.getDataSourceMap().isEmpty());
assertFalse(haDataSource.isInBlackList(name));
assertTrue(ds.isClosed());
}
@Test
public void testRemoveDataSources_nodeNotExisted() {
DataSource ds = new MockDataSource("foo");
haDataSource.getDataSourceMap().put("foo", ds);
updater.getNodesToDel().add("bar");
updater.removeDataSources();
assertFalse(updater.getNodesToDel().contains("bar"));
assertEquals(1, haDataSource.getDataSourceMap().size());
}
@Test
public void testRemoveDataSources_notDruid() {
DataSource ds = new MockDataSource("foo");
haDataSource.getDataSourceMap().put("foo", ds);
updater.getNodesToDel().add("foo");
updater.removeDataSources();
assertFalse(updater.getNodesToDel().contains("foo"));
assertTrue(haDataSource.getDataSourceMap().isEmpty());
}
@Test
public void testAddNode() {
String url = "jdbc:derby:memory:foo;create=true";
String name = "foo";
addNode(url, name);
String targetUrl = ((DruidDataSource) haDataSource.getDataSourceMap().get(name)).getUrl();
assertEquals(1, haDataSource.getDataSourceMap().size());
assertEquals(url, targetUrl);
}
@Test
public void testAddNode_alreadyExisted() {
DataSource ds = new MockDataSource("foo");
haDataSource.getDataSourceMap().put("foo", ds);
updater.getNodesToDel().add("foo");
addNode("foo_url", "foo");
assertFalse(updater.getNodesToDel().contains("foo"));
assertFalse(haDataSource.isInBlackList("foo"));
assertEquals(ds, haDataSource.getDataSourceMap().get("foo"));
}
@Test
public void testDeleteNode_notExisted() {
NodeEvent event = new NodeEvent();
event.setNodeName("foo");
event.setType(NodeEventType.DELETE);
updater.deleteNode(event);
assertTrue(updater.getNodesToDel().isEmpty());
assertFalse(haDataSource.isInBlackList("foo"));
}
@Test
public void testDeleteNode() {
haDataSource.getDataSourceMap().put("foo", new MockDataSource("foo"));
NodeEvent event = new NodeEvent();
event.setNodeName("foo");
event.setType(NodeEventType.DELETE);
updater.deleteNode(event);
validateDeleteNode();
}
private void validateDeleteNode() {
assertEquals(1, updater.getNodesToDel().size());
assertTrue(updater.getNodesToDel().contains("foo"));
assertTrue(haDataSource.isInBlackList("foo"));
}
private void addNode(String url, String name) {
NodeEvent event = new NodeEvent();
event.setNodeName(name);
event.setUrl(url);
event.setType(NodeEventType.ADD);
updater.addNode(event);
}
}
@@ -1,4 +1,8 @@
prefix1.foo.url=jdbc:derby:memory:foo1;create=true
prefix1.foo.username=foo
prefix1.foo.password=foo_password
prefix2.foo.url=jdbc:derby:memory:foo2;create=true
prefix2.foo.username=bar
prefix2.foo.password=bar_password
foo.url=jdbc:derby:memory:foo3;create=true
bar=jdbc:derby:memory:bar;create=true