dubbo源码学习笔记----monitor

核心类

public abstract class AbstractMonitorFactory implements MonitorFactory {
private static final Logger logger = LoggerFactory.getLogger(AbstractMonitorFactory.class);

// lock for getting monitor center
private static final ReentrantLock LOCK = new ReentrantLock();

// monitor centers Map<RegistryAddress, Registry>
private static final Map<String, Monitor> MONITORS = new ConcurrentHashMap<String, Monitor>();

private static final Map<String, ListenableFuture<Monitor>> FUTURES = new ConcurrentHashMap<String, ListenableFuture<Monitor>>();

private static final ExecutorService executor = new ThreadPoolExecutor(0, 10, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>(), new NamedThreadFactory("DubboMonitorCreator", true));

public static Collection<Monitor> getMonitors() {
return Collections.unmodifiableCollection(MONITORS.values());
}

针对于每个注册进来的URL有个对应的Monitor状态跟踪类，每个Monitor状态跟踪类，通过一个Listener进行绑定。

class MonitorListener implements Runnable {

private String key;

public MonitorListener(String key) {
this.key = key;
}

@Override
public void run() {
try {
ListenableFuture<Monitor> listenableFuture = AbstractMonitorFactory.FUTURES.get(key);
AbstractMonitorFactory.MONITORS.put(key, listenableFuture.get());
AbstractMonitorFactory.FUTURES.remove(key);
} catch (InterruptedException e) {
logger.warn("Thread was interrupted unexpectedly, monitor will never be got.");
AbstractMonitorFactory.FUTURES.remove(key);
} catch (ExecutionException e) {
logger.warn("Create monitor failed, monitor data will not be collected until you fix this problem. ", e);
}
}
}

将这个Listener交给一个线程池定时拉取Monitor信息。

LOCK.lock();
try {
monitor = MONITORS.get(key);
future = FUTURES.get(key);
if (monitor != null || future != null) {
return monitor;
}

final URL monitorUrl = url;
final ListenableFutureTask<Monitor> listenableFutureTask = ListenableFutureTask.create(new MonitorCreator(monitorUrl));
listenableFutureTask.addListener(new MonitorListener(key));
executor.execute(listenableFutureTask);
FUTURES.put(key, listenableFutureTask);

return null;
} finally {
// unlock
LOCK.unlock();
}

Monitor使用

<bean id="monitorService" class="com.alibaba.dubbo.monitor.simple.SimpleMonitorService">
</bean>

<dubbo:application name="${dubbo.application.name}" owner="${dubbo.application.owner}"/>

<dubbo:registry address="${dubbo.registry.address}"/>

<dubbo:protocol name="dubbo" port="${dubbo.protocol.port}"/>

<dubbo:service interface="com.alibaba.dubbo.monitor.MonitorService" ref="monitorService" delay="-1"/>

<dubbo:reference id="registryService" interface="com.alibaba.dubbo.registry.RegistryService"/>

Monitor信息应该是和注册中心信息放在一起，redis或是zookeeper。

容器container核心方法两个：

public interface Container {

/**
* start.
*/
void start();

/**
* stop.
*/
void stop();

}

代理方式收集monitor信息

public SimpleMonitorService() {
queue = new LinkedBlockingQueue<URL>(Integer.parseInt(ConfigUtils.getProperty("dubbo.monitor.queue", "100000")));
writeThread = new Thread(new Runnable() {
public void run() {
while (running) {
try {
write(); // write statistics
} catch (Throwable t) {
logger.error("Unexpected error occur at write stat log, cause: " + t.getMessage(), t);
try {
Thread.sleep(5000); // retry after 5 secs
} catch (Throwable t2) {
}
}
}
}
});
writeThread.setDaemon(true);
writeThread.setName("DubboMonitorAsyncWriteLogThread");
writeThread.start();
chartFuture = scheduledExecutorService.scheduleWithFixedDelay(new Runnable() {
public void run() {
try {
draw(); // draw chart
} catch (Throwable t) {
logger.error("Unexpected error occur at draw stat chart, cause: " + t.getMessage(), t);
}
}
}, 1, 300, TimeUnit.SECONDS);
statisticsDirectory = ConfigUtils.getProperty("dubbo.statistics.directory");
chartsDirectory = ConfigUtils.getProperty("dubbo.charts.directory");
}

起了一个后台线程写信息到本地文件，定时采集Monitor信息同步到统一Monitor中心。