聊聊nacos的ConfigDataChangeEvent
序
本文主要研究一下nacos的ConfigDataChangeEventjava
Event
nacos-1.1.3/config/src/main/java/com/alibaba/nacos/config/server/utils/event/EventDispatcher.javagit
public interface Event {
}
- Event是個空接口，僅僅用於標名事件
ConfigDataChangeEvent
nacos-1.1.3/config/src/main/java/com/alibaba/nacos/config/server/service/ConfigDataChangeEvent.javagithub
public class ConfigDataChangeEvent implements Event {
final public boolean isBeta;
final public String dataId;
final public String group;
final public String tenant;
final public String tag;
final public long lastModifiedTs;
public ConfigDataChangeEvent(String dataId, String group, long gmtModified) {
this(false, dataId, group, gmtModified);
}
public ConfigDataChangeEvent(boolean isBeta, String dataId, String group, String tenant, long gmtModified) {
if (null == dataId || null == group) {
throw new IllegalArgumentException();
}
this.isBeta = isBeta;
this.dataId = dataId;
this.group = group;
this.tenant = tenant;
this.tag = null;
this.lastModifiedTs = gmtModified;
}
public ConfigDataChangeEvent(boolean isBeta, String dataId, String group, long gmtModified) {
this(isBeta, dataId, group, StringUtils.EMPTY, gmtModified);
}
public ConfigDataChangeEvent(boolean isBeta, String dataId, String group, String tenant, String tag,
long gmtModified) {
if (null == dataId || null == group) {
throw new IllegalArgumentException();
}
this.isBeta = isBeta;
this.dataId = dataId;
this.group = group;
this.tenant = tenant;
this.tag = tag;
this.lastModifiedTs = gmtModified;
}
}
- ConfigDataChangeEvent實現了Event接口，它包含有isBeta、dataId、group、tenant、tag、lastModifiedTs屬性
AbstractEventListener
nacos-1.1.3/config/src/main/java/com/alibaba/nacos/config/server/utils/event/EventDispatcher.java併發
static public abstract class AbstractEventListener {
public AbstractEventListener() {
/**
* automatic register
*/
EventDispatcher.addEventListener(this);
}
/**
* 感興趣的事件列表
*
* @return event list
*/
abstract public List<Class<? extends Event>> interest();
/**
* 處理事件
*
* @param event event
*/
abstract public void onEvent(Event event);
}
- AbstractEventListener接口定義了interest、onEvent兩個抽象方法
AsyncNotifyService
nacos-1.1.3/config/src/main/java/com/alibaba/nacos/config/server/service/notify/AsyncNotifyService.javaasync
@Service
public class AsyncNotifyService extends AbstractEventListener {
@Override
public List<Class<? extends Event>> interest() {
List<Class<? extends Event>> types = new ArrayList<Class<? extends Event>>();
// 觸發配置變動同步通知
types.add(ConfigDataChangeEvent.class);
return types;
}
@Override
public void onEvent(Event event) {
// 併發產生 ConfigDataChangeEvent
if (event instanceof ConfigDataChangeEvent) {
ConfigDataChangeEvent evt = (ConfigDataChangeEvent) event;
long dumpTs = evt.lastModifiedTs;
String dataId = evt.dataId;
String group = evt.group;
String tenant = evt.tenant;
String tag = evt.tag;
List<?> ipList = serverListService.getServerList();
// 其實這裏任何類型隊列均可以
Queue<NotifySingleTask> queue = new LinkedList<NotifySingleTask>();
for (int i = 0; i < ipList.size(); i++) {
queue.add(new NotifySingleTask(dataId, group, tenant, tag, dumpTs, (String) ipList.get(i), evt.isBeta));
}
EXECUTOR.execute(new AsyncTask(httpclient, queue));
}
}
@Autowired
public AsyncNotifyService(ServerListService serverListService) {
this.serverListService = serverListService;
httpclient.start();
}
//......
}
- AsyncNotifyService繼承了AbstractEventListener接口，其interest返回的列表包含了ConfigDataChangeEvent；其onEvent方法判斷是ConfigDataChangeEvent，則會經過serverListService.getServerList()獲取ipList，而後遍歷該列表往queue添加NotifySingleTask，而後使用該queue建立AsyncTask並提交線程池執行
AsyncTask
nacos-1.1.3/config/src/main/java/com/alibaba/nacos/config/server/service/notify/AsyncNotifyService.javaide
class AsyncTask implements Runnable {
public AsyncTask(CloseableHttpAsyncClient httpclient, Queue<NotifySingleTask> queue) {
this.httpclient = httpclient;
this.queue = queue;
}
@Override
public void run() {
executeAsyncInvoke();
}
private void executeAsyncInvoke() {
while (!queue.isEmpty()) {
NotifySingleTask task = queue.poll();
String targetIp = task.getTargetIP();
if (serverListService.getServerList().contains(
targetIp)) {
// 啓動健康檢查且有不監控的ip則直接把放到通知隊列，不然通知
if (serverListService.isHealthCheck()
&& ServerListService.getServerListUnhealth().contains(targetIp)) {
// target ip 不健康，則放入通知列表中
ConfigTraceService.logNotifyEvent(task.getDataId(), task.getGroup(), task.getTenant(), null,
task.getLastModified(),
LOCAL_IP, ConfigTraceService.NOTIFY_EVENT_UNHEALTH, 0, task.target);
// get delay time and set fail count to the task
asyncTaskExecute(task);
} else {
HttpGet request = new HttpGet(task.url);
request.setHeader(NotifyService.NOTIFY_HEADER_LAST_MODIFIED,
String.valueOf(task.getLastModified()));
request.setHeader(NotifyService.NOTIFY_HEADER_OP_HANDLE_IP, LOCAL_IP);
if (task.isBeta) {
request.setHeader("isBeta", "true");
}
httpclient.execute(request, new AsyncNotifyCallBack(httpclient, task));
}
}
}
}
private Queue<NotifySingleTask> queue;
private CloseableHttpAsyncClient httpclient;
}
- AsyncTask實現了Runnable接口，其run方法執行executeAsyncInvoke方法；該方法會不斷從queue取出NotifySingleTask，而後判斷目標ip是否健康不健康則延遲再次執行該task；若是是健康的話則使用httpclient向目標地址發送get請求，而後註冊AsyncNotifyCallBack
AsyncNotifyCallBack
nacos-1.1.3/config/src/main/java/com/alibaba/nacos/config/server/service/notify/AsyncNotifyService.javaui
class AsyncNotifyCallBack implements FutureCallback<HttpResponse> {
public AsyncNotifyCallBack(CloseableHttpAsyncClient httpClient, NotifySingleTask task) {
this.task = task;
this.httpClient = httpClient;
}
@Override
public void completed(HttpResponse response) {
long delayed = System.currentTimeMillis() - task.getLastModified();
if (response.getStatusLine().getStatusCode() == HttpStatus.SC_OK) {
ConfigTraceService.logNotifyEvent(task.getDataId(),
task.getGroup(), task.getTenant(), null, task.getLastModified(),
LOCAL_IP,
ConfigTraceService.NOTIFY_EVENT_OK, delayed,
task.target);
} else {
log.error("[notify-error] target:{} dataId:{} group:{} ts:{} code:{}",
task.target, task.getDataId(), task.getGroup(), task.getLastModified(), response.getStatusLine().getStatusCode());
ConfigTraceService.logNotifyEvent(task.getDataId(),
task.getGroup(), task.getTenant(), null, task.getLastModified(),
LOCAL_IP,
ConfigTraceService.NOTIFY_EVENT_ERROR, delayed,
task.target);
//get delay time and set fail count to the task
asyncTaskExecute(task);
LogUtil.notifyLog.error("[notify-retry] target:{} dataId:{} group:{} ts:{}",
task.target, task.getDataId(), task.getGroup(), task.getLastModified());
MetricsMonitor.getConfigNotifyException().increment();
}
HttpClientUtils.closeQuietly(response);
}
@Override
public void failed(Exception ex) {
long delayed = System.currentTimeMillis() - task.getLastModified();
log.error("[notify-exception] target:{} dataId:{} group:{} ts:{} ex:{}",
task.target, task.getDataId(), task.getGroup(), task.getLastModified(), ex.toString());
ConfigTraceService.logNotifyEvent(task.getDataId(),
task.getGroup(), task.getTenant(), null, task.getLastModified(),
LOCAL_IP,
ConfigTraceService.NOTIFY_EVENT_EXCEPTION, delayed,
task.target);
//get delay time and set fail count to the task
asyncTaskExecute(task);
LogUtil.notifyLog.error("[notify-retry] target:{} dataId:{} group:{} ts:{}",
task.target, task.getDataId(), task.getGroup(), task.getLastModified());
MetricsMonitor.getConfigNotifyException().increment();
}
@Override
public void cancelled() {
LogUtil.notifyLog.error("[notify-exception] target:{} dataId:{} group:{} ts:{} method:{}",
task.target, task.getDataId(), task.getGroup(), task.getLastModified(), "CANCELED");
//get delay time and set fail count to the task
asyncTaskExecute(task);
LogUtil.notifyLog.error("[notify-retry] target:{} dataId:{} group:{} ts:{}",
task.target, task.getDataId(), task.getGroup(), task.getLastModified());
MetricsMonitor.getConfigNotifyException().increment();
}
private NotifySingleTask task;
private CloseableHttpAsyncClient httpClient;
}
- AsyncNotifyCallBack實現了FutureCallback接口，其completed方法判斷請求是不是HttpStatus.SC_OK，不是的話會再次延遲調度該任務；其failed、cancelled方法也是會再次延時調度該任務
小結
ConfigDataChangeEvent實現了Event接口，它包含有isBeta、dataId、group、tenant、tag、lastModifiedTs屬性；AsyncNotifyService繼承了AbstractEventListener接口，其interest返回的列表包含了ConfigDataChangeEvent；其onEvent方法判斷是ConfigDataChangeEvent，則會經過serverListService.getServerList()獲取ipList，而後遍歷該列表往queue添加NotifySingleTask，而後使用該queue建立AsyncTask並提交線程池執行this
doc
相關文章
- 1. 聊聊nacos的ConfigDataChangeEvent
- 2. 聊聊nacos的ServerStatusManager
- 3. 聊聊nacos的ServiceReporter
- 4. 聊聊nacos的HttpHealthCheckProcessor
- 5. 聊聊nacos的notifyConfigInfo
- 6. 聊聊nacos的NacosDiscoveryAutoConfiguration
- 7. 聊聊nacos的DistroFilter
- 8. 聊聊nacos的DelegateConsistencyServiceImpl
- 9. 聊聊nacos的MysqlHealthCheckProcessor
- 10. 聊聊nacos的RaftProxy
- 更多相關文章...
- • PHP imagecolorclosest - 取得與指定的顏色最接近的顏色的索引值 - PHP參考手冊
- • PHP imagecolorclosesthwb - 取得與指定的顏色最接近的色度的黑白色的索引 - PHP參考手冊
- • 漫談MySQL的鎖機制
- • 互聯網組織的未來：剖析GitHub員工的任性之源
相關標籤/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
最新文章
- 1. Window下Ribbit MQ安裝
- 2. Linux下Redis安裝及集羣搭建
- 3. shiny搭建網站填坑戰略
- 4. Mysql8.0.22安裝與配置詳細教程
- 5. Hadoop安裝及配置
- 6. Python爬蟲初學筆記
- 7. 部署LVS-Keepalived高可用集羣
- 8. keepalived+mysql高可用集羣
- 9. jenkins 公鑰配置
- 10. HA實用詳解
歡迎關注本站公眾號,獲取更多信息
相關文章
- 1. 聊聊nacos的ConfigDataChangeEvent
- 2. 聊聊nacos的ServerStatusManager
- 3. 聊聊nacos的ServiceReporter
- 4. 聊聊nacos的HttpHealthCheckProcessor
- 5. 聊聊nacos的notifyConfigInfo
- 6. 聊聊nacos的NacosDiscoveryAutoConfiguration
- 7. 聊聊nacos的DistroFilter
- 8. 聊聊nacos的DelegateConsistencyServiceImpl
- 9. 聊聊nacos的MysqlHealthCheckProcessor
- 10. 聊聊nacos的RaftProxy