SofaRpcMetrics extends Subscriber implements MeterBinder, AutoCloseable { @Override public void bindTo(MeterRegistry registry) { Gauge.builder("sofa.threadpool.config.core", () -> Optional.of(serverConfig) .map(AtomicReference::get) .map(ServerConfig::getCoreThreads) .orElse(0)) .tags(common) .baseUnit(BaseUnits.THREADS) .register(registry); Gauge.builder("sofa.threadpool.config.max", () -> Optional.of(serverConfig) .map(AtomicReference::get) .map(ServerConfig::getMaxThreads) .orElse(0)) .tags(common) .baseUnit(BaseUnits.THREADS) .register(registry); Gauge.builder("sofa.threadpool.config.queue", () -> Optional.of(serverConfig) .map(AtomicReference::get) .map(ServerConfig::getQueues) .orElse(0)) .tags(common) .baseUnit(BaseUnits.TASKS) .register(registry); Gauge.builder("sofa.threadpool.active", () -> Optional.of(executor) .map(AtomicReference::get) .map(ThreadPoolExecutor::getActiveCount) .orElse(0)) .tags(common) .baseUnit(BaseUnits.THREADS) .register(registry); Gauge.builder("sofa.threadpool.idle", () -> Optional.of(executor) .map(AtomicReference::get) .map(e -> e.getPoolSize() - e.getActiveCount()) .orElse(0)) .tags(common) .baseUnit(BaseUnits.THREADS) .register(registry); Gauge.builder("sofa.threadpool.queue.size", () -> Optional.of(executor) .map(AtomicReference::get) .map(ThreadPoolExecutor::getQueue) .map(Collection::size) .orElse(0)) .tags(common) .baseUnit(BaseUnits.TASKS) .register(registry); provider = Counter.builder("sofa.provider") .tags(common) .register(registry); consumer = Counter.builder("sofa.consumer") .tags(common) .register(registry); initialed.set(registry); } SofaRpcMetrics();  SofaRpcMetrics(Iterable<Tag> common); @Override void bindTo(MeterRegistry registry); @Override void onEvent(Event event); @Override void close();  }
@Test public void testMicrometerMetrics() throws NoSuchMethodException, InvocationTargetException, IllegalAccessException { SimpleMeterRegistry registry = new SimpleMeterRegistry(); try (SofaRpcMetrics metrics = new SofaRpcMetrics()) { metrics.bindTo(registry); Method handleEvent = EventBus.class.getDeclaredMethod( "handleEvent", Subscriber.class, Event.class); handleEvent.setAccessible(true); SofaRequest request = buildRequest(); SofaResponse response = buildResponse(); RpcInternalContext.getContext() .setAttachment(RpcConstants.INTERNAL_KEY_CLIENT_ELAPSE, 100) .setAttachment(RpcConstants.INTERNAL_KEY_IMPL_ELAPSE, 10) .setAttachment(RpcConstants.INTERNAL_KEY_REQ_SIZE, 3) .setAttachment(RpcConstants.INTERNAL_KEY_RESP_SIZE, 4); handleEvent.invoke(EventBus.class, metrics, new ClientEndInvokeEvent(request, response, null)); handleEvent.invoke(EventBus.class, metrics, new ServerSendEvent(request, response, null)); ServerConfig serverConfig = new ServerConfig(); handleEvent.invoke(EventBus.class, metrics, new ServerStartedEvent(serverConfig, new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>()))); handleEvent.invoke(EventBus.class, metrics, new ServerStoppedEvent(serverConfig)); handleEvent.invoke(EventBus.class, metrics, new ProviderPubEvent(new ProviderConfig<>())); handleEvent.invoke(EventBus.class, metrics, new ConsumerSubEvent(new ConsumerConfig<>())); Assert.assertEquals(12, registry.getMeters().size()); } }