The basic principle of Dubbo monitoring is to collect service call concurrency and service response time during service invocation, and then report statistical data to the monitoring center at a certain frequency. 1, source analysis MonitorFilter
- The filter is used to monitor the filter and report service invocation data to the monitoring center.
- Usage Scenario Setting up a monitoring center to monitor Dubbo service invocation.
- Block condition Non-block filter.
1.1 MonitorFilter statement
/** * MonitorFilter. (SPI, Singleton, ThreadSafe) */ @Activate(group = {Constants.PROVIDER, Constants.CONSUMER}) public class MonitorFilter implements Filter {//Copy the codeNote: MonitorFilter takes effect at both producer and consumer ends. 1.2 Explain the getConcurrent method
// concurrent counter private AtomicInteger getConcurrent(Invoker<? > invoker, Invocation invocation) { String key = invoker.getInterface().getName() + "." + invocation.getMethodName(); // @1 AtomicInteger concurrent = concurrents.get(key); if (concurrent == null) { concurrents.putIfAbsent(key, new AtomicInteger()); // @2 concurrent = concurrents.get(key); } return concurrent; }Copy the codeBasically, get the call-count calculator for the currently invoked service. @1: ConcurrentMap< String, AtomicInteger > is used as the cache container with key: interfaceName + “. + methodName. Code @2: Creates AtomicInteger if it is called for the first time, otherwise returns the original counter. 1.3 Invoker method in detail
@Override
public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
if (invoker.getUrl().hasParameter(Constants.MONITOR_KEY)) { // @1
RpcContext context = RpcContext.getContext(); // provider must fetch context before invoke() gets called // @2
String remoteHost = context.getRemoteHost();
long start = System.currentTimeMillis(); // record start timestamp
getConcurrent(invoker, invocation).incrementAndGet(); // count up // @3
try {
Result result = invoker.invoke(invocation); // proceed invocation chain // @4
collect(invoker, invocation, result, remoteHost, start, false); // @5
return result;
} catch (RpcException e) {
collect(invoker, invocation, null, remoteHost, start, true); // @6
throw e;
} finally {
getConcurrent(invoker, invocation).decrementAndGet(); // count down // @7
}
} else {
return invoker.invoke(invocation);
}
}
Copy the codeCode @1: If monitor is present in the URL, a monitoring center is set up to collect call information. Code @2: Gets the context of this service invocation. Code @3: The number of concurrent service invocations increases by 1, (not the total number of service invocations, but the concurrent invocation of the current service). Code @4: Log the current time before executing the method, and then call the next filter until the real service is called. Code @5: Call the collect method to collect the call information. Code @6: Collect error information if the call sends RPC incorrectly. Code @7: One service call ends, the number of concurrent calls is reduced by one. Next, analyze the collect method. 1.4 Invoker method details
// collect info private void collect(Invoker<? > invoker, Invocation invocation, Result result, String remoteHost, long start, boolean error) { // @1 try { // ---- service statistics ---- // @2 start long elapsed = System.currentTimeMillis() - start; // invocation cost int concurrent = getConcurrent(invoker, invocation).get(); // current concurrent count String application = invoker.getUrl().getParameter(Constants.APPLICATION_KEY); String service = invoker.getInterface().getName(); // service name String method = RpcUtils.getMethodName(invocation); // method name String group = invoker.getUrl().getParameter(Constants.GROUP_KEY); String version = invoker.getUrl().getParameter(Constants.VERSION_KEY); URL url = invoker.getUrl().getUrlParameter(Constants.MONITOR_KEY); // @2 end Monitor monitor = monitorFactory.getMonitor(url); // @3 if (monitor == null) { return; } int localPort; String remoteKey; String remoteValue; if (Constants.CONSUMER_SIDE.equals(invoker.getUrl().getParameter(Constants.SIDE_KEY))) { // @4 // ---- for service consumer ---- localPort = 0; remoteKey = MonitorService.PROVIDER; remoteValue = invoker.getUrl().getAddress(); } else { // @5 // ---- for service provider ---- localPort = invoker.getUrl().getPort(); remoteKey = MonitorService.CONSUMER; remoteValue = remoteHost; } String input = "", output = ""; if (invocation.getAttachment(Constants.INPUT_KEY) ! = null) { // @6 input = invocation.getAttachment(Constants.INPUT_KEY); } if (result ! = null && result.getAttachment(Constants.OUTPUT_KEY) ! = null) { // @7 output = result.getAttachment(Constants.OUTPUT_KEY); } monitor.collect(new URL(Constants.COUNT_PROTOCOL, // @8 NetUtils.getLocalHost(), localPort, service + "/" + method, MonitorService.APPLICATION, application, MonitorService.INTERFACE, service, MonitorService.METHOD, method, remoteKey, remoteValue, error ? MonitorService.FAILURE : MonitorService.SUCCESS, "1", MonitorService.ELAPSED, String.valueOf(elapsed), MonitorService.CONCURRENT, String.valueOf(concurrent), Constants.INPUT_KEY, input, Constants.OUTPUT_KEY, output, Constants.GROUP_KEY, group, Constants.VERSION_KEY, version)); } catch (Throwable t) { logger.error("Failed to monitor count service " + invoker.getUrl() + ", cause: " + t.getMessage(), t); }}Copy the codeCode @1: Parameter description.
Invoker< ? > invoker: Service invoker.
Invocation: Information about this service Invocation
Result Result: Execution Result
String remoteHost: Caller host information.
Long start: time when the service starts to invoke.
Boolean error: Indicates whether an error has occurred.
Code @2: Basic statistical information
Elapsed: The time for the service invocation.
Concurrent: indicates the current concurrency. (Current number of concurrent service invocations).
Application: indicates the name of the service owning application.
Service: indicates the service name.
Method: indicates the name of the method.
Group: indicates the group to which the service belongs.
Version: indicates the service version number
URL URL: indicates the URL of the monitoring center.
Code @ 3: according to the monitoring center for monitoring implementation class, this is the monitoring center to realize the extension point, default com. Alibaba. Dubbo. Monitor. The dubbo. DubboMonitor.
@4: If it is on the consumer side, since Monitor works on both the consumer side and the server side:
LocalPort: set the localPort to 0.
RemoteKey: MonitorService.PROVIDER, which represents the server.
RemoteValue: invoker.geturl ().getAddress(), whose value is (registry address) or service provider address (client directly connected to server).
@5: If it is a server:
LocalPort: indicates the service port number of the server.
RemoteKey: MonitorService.CONSUMER, indicating that the remote is the service CONSUMER.
RemoteValue: consumer host(IP :port).
Code @6: Get the number of bytes of this service invocation request packet, which will be decoded in the RpcContext on the server.
@7: Get the number of bytes of the response package for this service invocation, which will be written when the server encodes the response package. For the specific code, refer to the DubboCountCodec class.
Code @8: Call Monitor # Collect to collect call information. Monitor implements DubboMonitor by default. The protocol used to count: / / localhost: localPort/service/method? Application = applicationName&remoteKey = remoteValue&success | failure = 1 & elapsed = call overhead & concurrent = number of concurrent calls & input = the number of bytes refs & output = response Bytes &group= Group to which the service belongs &version= version.
2, source code analysis DubboMonitor implementation principle
The default Monitor implementation class in Dubbo is DubboMonitor:
Introduction to core attributes:
- private final ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(3, New NamedThreadFactory(” DubboMonitorSendTimer “, true) : Scheduled thread pool, using a thread pool of 3 threads, thread name is DubboMonitorSendTimer.
- private final ScheduledFuture< ? > sendFuture: Scheduling task future. Private final Invoker< MonitorService > monitorInvoker: Monitoring scheduling Invoker. The monitoring center in Dubbo will expose the service as a service provider. Service providers and service consumers can subscribe to the service through the registry and report call statistics to the monitoring center through the Invoker, that is, one report is one Dubbo RPC service call. The implementation class is DubboInvoker, through which the remote Monitor service can be invoked using the Dubbo protocol.
- Private Final MonitorService Proxy proxies for monitorInvoker, primarily toString, Hashcode, and equals, do not initiate calls to the MonitorServer service provider via RPC. AbstractProxyFactory#getProxy; JavassistProxyFactory; Com. Alibaba. Dubbo. RPC. Proxy. InvokerInvocationHandler# invoke.
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { String methodName = method.getName(); Class<? >[] parameterTypes = method.getParameterTypes(); if (method.getDeclaringClass() == Object.class) { return method.invoke(invoker, args); } if ("toString".equals(methodName) && parameterTypes.length == 0) { return invoker.toString(); } if ("hashCode".equals(methodName) && parameterTypes.length == 0) { return invoker.hashCode(); } if ("equals".equals(methodName) && parameterTypes.length == 1) { return invoker.equals(args[0]); } return invoker.invoke(new RpcInvocation(method, args)).recreate(); }Copy the code- Private Final Long monitorInterval: Frequency of reporting to the monitoring center, that is, frequency of calling RPC services of the MonitorService. The default value is 1 minute.
- Private Final ConcurrentMap< Statistics, AtomicReference< Long []>> statisticsMap: Statistics Map.
2.1 Constructor analysis
public DubboMonitor(Invoker<MonitorService> monitorInvoker, MonitorService monitorService) {
this.monitorInvoker = monitorInvoker;
this.monitorService = monitorService;
this.monitorInterval = monitorInvoker.getUrl().getPositiveParameter("interval", 60000); // @1
// collect timer for collecting statistics data
sendFuture = scheduledExecutorService.scheduleWithFixedDelay(new Runnable() { // @2
@Override
public void run() {
// collect data
try {
send();
} catch (Throwable t) {
logger.error("Unexpected error occur at send statistic, cause: " + t.getMessage(), t);
}
}
}, monitorInterval, monitorInterval, TimeUnit.MILLISECONDS);
}
Copy the codeThe code @1 gets the interval attribute from the URL parameter. If it is empty, the default is 60000, representing 60S. Code @2: Start the scheduled scheduling task, execute the send() method every 60 seconds by default, and report service call statistics to the monitoring center. 2.2 Collect Methods for collecting statistics
public void collect(URL url) { // data to collect from url int success = url.getParameter(MonitorService.SUCCESS, 0); int failure = url.getParameter(MonitorService.FAILURE, 0); int input = url.getParameter(MonitorService.INPUT, 0); int output = url.getParameter(MonitorService.OUTPUT, 0); int elapsed = url.getParameter(MonitorService.ELAPSED, 0); int concurrent = url.getParameter(MonitorService.CONCURRENT, 0); // init atomic reference Statistics statistics = new Statistics(url); AtomicReference<long[]> reference = statisticsMap.get(statistics); if (reference == null) { statisticsMap.putIfAbsent(statistics, new AtomicReference<long[]>()); reference = statisticsMap.get(statistics); } // use CompareAndSet to sum long[] current; long[] update = new long[LENGTH]; do { current = reference.get(); if (current == null) { update[0] = success; update[1] = failure; update[2] = input; update[3] = output; update[4] = elapsed; update[5] = concurrent; update[6] = input; update[7] = output; update[8] = elapsed; update[9] = concurrent; } else { update[0] = current[0] + success; update[1] = current[1] + failure; update[2] = current[2] + input; update[3] = current[3] + output; update[4] = current[4] + elapsed; update[5] = (current[5] + concurrent) / 2; update[6] = current[6] > input ? current[6] : input; update[7] = current[7] > output ? current[7] : output; update[8] = current[8] > elapsed ? current[8] : elapsed; update[9] = current[9] > concurrent ? current[9] : concurrent; } } while (! reference.compareAndSet(current, update)); }Copy the codeThe collected information consists of 10 fields. Update [0] : number of successful calls update[1] : number of failed calls UPDATE [2] : Total call traffic (total size of request packets). Update [3] : Total response traffic (total size of response packets). Update [4] : Total response time (total service invocation overhead). Update [5] : Average TPS of a collection cycle. Update [6] : Maximum request packet size. Update [7] : Maximum response packet size. Update [8] : Maximum response time. Update [9] : Maximum TPS.
2.3 Send Uses monitorService and monitorInvoker to call RPC service to report data to the monitoring center. Let’s look at the implementation of the monitoring center.
3. Implementation principle of Dubbo Monitoring center
Dubbo officially offers a simplified version of the monitoring center, with the project dubbo-OPS: Dubbo-monitor-Simple. The project is a Spring-boot project, you can see the background management interface after starting.
The project service provider file is as follows:
It can be seen that the monitoring center service provider implementation class is SimpleMonitorService and its implementation interface is MonitorService.
Next, the implementation of SimpleMonitorService monitoring center is analyzed, focusing on the following two points:
1. Monitoring data persistence.
2. Monitor report generation logic.
Description of core attributes:
- ScheduledExecutorService ScheduledExecutorService: schedules a thread that writes monitoring data to a scheduled task in the pie chart. This thread is fixed.
- Thread writeThread: Monitors data persistence threads.
- BlockingQueue< URL > queue: persistent data task BlockingQueue.
- String statisticsDirectory = “Statistics” : Data persistence directory. SimpleMonitorService persists data to a disk file. This value specifies the directory name.
- String chartsDirectory = “Charts” : Indicates the chart storage directory.
- Private Volatile Boolean RUNNING = true: Specifies whether the persistent data thread is running.
3.1 SimpleMonitorService constructor
public SimpleMonitorService() { queue = new LinkedBlockingQueue<URL>(Integer.parseInt(ConfigUtils.getProperty("dubbo.monitor.queue", "100000"))); // @1 writeThread = new Thread(new Runnable() { // @2 start 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(); // @2 end 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); // @3 statisticsDirectory = ConfigUtils.getProperty("dubbo.statistics.directory"); chartsDirectory = ConfigUtils.getProperty("dubbo.charts.directory"); / / @ 4}Copy the code@1: Create LinkedBlockingQueue. The default size is 100000. Change the default value by configuring dubo.monitor. queue. Code @ 2: create a persistent monitoring data threads, called DubboMonitorAsyncWriteLogThread, its mission for monitoring the raw data from LinkedBlockingQueue, no data is blocked, if the queue and then written to the file. Code @3: Enable scheduled scheduling tasks, and generate pie charts according to persistent monitoring data every 5 minutes. Code @4: Get data persistence directory and pie chart directory. 3.2 SimpleMonitorService# write
private void write() throws Exception {
URL statistics = queue.take();
if (POISON_PROTOCOL.equals(statistics.getProtocol())) {
return;
}
String timestamp = statistics.getParameter(Constants.TIMESTAMP_KEY);
Date now;
if (timestamp == null || timestamp.length() == 0) {
now = new Date();
} else if (timestamp.length() == "yyyyMMddHHmmss".length()) {
now = new SimpleDateFormat("yyyyMMddHHmmss").parse(timestamp);
} else {
now = new Date(Long.parseLong(timestamp));
}
String day = new SimpleDateFormat("yyyyMMdd").format(now);
SimpleDateFormat format = new SimpleDateFormat("HHmm");
for (String key : types) {
try {
String type;
String consumer;
String provider;
if (statistics.hasParameter(PROVIDER)) {
type = CONSUMER;
consumer = statistics.getHost();
provider = statistics.getParameter(PROVIDER);
int i = provider.indexOf(':');
if (i > 0) {
provider = provider.substring(0, i);
}
} else {
type = PROVIDER;
consumer = statistics.getParameter(CONSUMER);
int i = consumer == null ? -1 : consumer.indexOf(':');
if (i > 0) {
consumer = consumer.substring(0, i);
}
provider = statistics.getHost();
}
String filename = statisticsDirectory
+ "/" + day
+ "/" + statistics.getServiceInterface()
+ "/" + statistics.getParameter(METHOD)
+ "/" + consumer
+ "/" + provider
+ "/" + type + "." + key;
File file = new File(filename);
File dir = file.getParentFile();
if (dir != null && !dir.exists()) {
dir.mkdirs();
}
FileWriter writer = new FileWriter(file, true);
try {
writer.write(format.format(now) + " " + statistics.getParameter(key, 0) + "\n");
writer.flush();
} finally {
writer.close();
}
} catch (Throwable t) {
logger.error(t.getMessage(), t);
}
}
}
Copy the codeData is stored on a physical disk, and its files are: ” d u b b o . s t a t i s t i c s . d i r e c t o r y / {dubbo.statistics.directory} / dubbo.statistics.directory/{day}/ F a c e I n t e r n a m e / {interfacename} / interfacename / / ${method} {consumer} / ${provider} / [consume | provider] / key “,
Key :{SUCCESS, FAILURE, ELAPSED, CONCURRENT, MAX_ELAPSED, MAX_CONCURRENT}, respectively, count of ELAPSED, count of ELAPSED, count of ELAPSED, count of ELAPSED, TPS, Max ELAPSED, Max TPS} Its file storage is as follows:
Take provider.concurrent as an example to illustrate its contents:
The content is organized as follows: time (time: collected value).
3.3 the draw
According to the persistent data, create a pie chart in a specific directory, create a pie chart method createChart, specific use of JFreeChart related class diagram, here is not detailed explained, interested friends can baidu query related usage.
3.4 Monitoring center effect overview
3.4.1 Application Overview
This function describes the relationship between systems.
Table Field Description:
1, Application Name: Application Name
Providers: The app contains information about service Providers. Click there to see the URL of the service provider.
3, Consumers(1) : The application contains service Consumers information, click in to view the specific URL of service Consumers.
4, Depends On the application.
5, Used By: This application is Used By lazy.
3.4.2 Service List
Table Field Description:
Service Name: indicates the Service Name.
Application: indicates the name of the Application to which the service belongs.
Providers: information about service Providers. Click there to see the detailed information about service Providers.
Consumers: Consumer information of the service.
Statistics: indicates table Statistics
“Charts” : Indicates the pie chart statistics
Pie chart statistics are displayed in two dimensions :QPS(number of interface requests per second) and average response time (including maximum and minimum response time).
1. Install Dubbo Simple Monitoring center. Download dubbo-simple-Monitor from github Dubbo repository. Dubbo service provider Dubbo service provider Dubbo service provider Dubbo service provider Dubbo service provider Dubbo service provider Dubbo service provider Dubbo service provider Dubbo service provider Dubbo service provider Dubbo service provider Dubbo service provider Dubbo service provider Dubbo service provider Dubbo service provider Dubbo service provider Dubbo service provider Dubbo service provider Dubbo service provider Dubbo service provider Dubbo service provider Dubbo service provider Dubbo service provider Dubbo service provider Dubbo service provider Dubbo < dubbo: Monitor protocol= “Registry” />, which indicates that the address of the monitoring center is discovered from the registry and the service invocation information is submitted to the monitoring center. By default, the service provider invokes the Dubbo service of the monitoring center at a frequency of one minute (configurable) and reports the service invocation information to the monitoring center. The outage of the monitoring center does not affect the normal work of consumers and service providers. If you want to configure the call frequency, you can use the following configuration. By default, it is recommended to keep the frequency of one minute or even lower. If the frequency is set low, the pressure on the whole server will not increase further
< dubbo:monitor protocol="registry"> <dubbo:parameter key = "interval" value="60000"> <! </ dubbo:monitor>Copy the codeNote: Dubbo monitoring center, service provider, service consumer can be configured separately on demand.
Welcome to add the author micro signal (DINGwPMZ), add group discussion, the author quality column catalog: 1, source analysis RocketMQ column (40 +) 2, source analysis Sentinel column (12 +) 3, source analysis Dubbo column (28 +) 4, source analysis Mybatis column 5, source analysis Netty column (18 +) 6, source analysis JUC column Source code analysis (MyCat) for Elasticjob