com.codahale.metrics.graphite.Graphite#io.vertx.core.VertxOptions源码实例Demo

/**
 * The service entry point.
 */
public static void main(String[] arguments) {
  Configurator.initialize("default", CONSOLE_LOG_CONFIG);
  final ConfigStoreOptions fileStore = new ConfigStoreOptions().setType("file").setConfig(new JsonObject().put("path", "config.json"));
  final ConfigStoreOptions envConfig = new ConfigStoreOptions().setType("env");
  final ConfigStoreOptions sysConfig = new ConfigStoreOptions().setType("sys");
  final ConfigRetrieverOptions options = new ConfigRetrieverOptions().addStore(fileStore).addStore(envConfig).addStore(sysConfig);
  boolean debug = Arrays.asList(arguments).contains("--debug");

  final VertxOptions vertxOptions = new VertxOptions()
    .setWorkerPoolSize(NumberUtils.toInt(System.getenv(VERTX_WORKER_POOL_SIZE), 128))
    .setPreferNativeTransport(true);

  if (debug) {
    vertxOptions
        .setBlockedThreadCheckInterval(TimeUnit.MINUTES.toMillis(1))
        .setMaxEventLoopExecuteTime(TimeUnit.MINUTES.toMillis(1))
        .setMaxWorkerExecuteTime(TimeUnit.MINUTES.toMillis(1))
        .setWarningExceptionTime(TimeUnit.MINUTES.toMillis(1));
  }

  vertx = Vertx.vertx(vertxOptions);
  webClient = WebClient.create(Service.vertx, new WebClientOptions().setUserAgent(XYZ_HUB_USER_AGENT));
  ConfigRetriever retriever = ConfigRetriever.create(vertx, options);
  retriever.getConfig(Service::onConfigLoaded);
}
 

@BeforeClass
public static void setUp(TestContext testContext) {
    configuration = new InferenceConfiguration()
            .protocol(ServerProtocol.HTTP)
            .pipeline(SequencePipeline.builder()
                    .add(new LoggingStep().log(LoggingStep.Log.KEYS_AND_VALUES).logLevel(Level.ERROR))
                    .add(new MetricsTestingStep())
                    .build());

    Async async = testContext.async();

    vertx = DeployKonduitServing.deploy(new VertxOptions(),
            new DeploymentOptions(),
            configuration,
            handler -> {
                if(handler.succeeded()) {
                    inferenceDeploymentResult = handler.result();
                    async.complete();
                } else {
                    testContext.fail(handler.cause());
                }
            });
}
 

@BeforeClass
public static void setUp(TestContext testContext) {
    configuration = new InferenceConfiguration()
            .protocol(ServerProtocol.HTTP)
            .pipeline(SequencePipeline.builder()
                    .add(new LoggingStep().log(LoggingStep.Log.KEYS_AND_VALUES).logLevel(Level.ERROR))
                    .build())
            .customEndpoints(Collections.singletonList(CustomHttpEndpoint.class.getName()));

    Async async = testContext.async();

    vertx = DeployKonduitServing.deploy(new VertxOptions(),
            new DeploymentOptions(),
            configuration,
            handler -> {
                if(handler.succeeded()) {
                    inferenceDeploymentResult = handler.result();
                    async.complete();
                } else {
                    testContext.fail(handler.cause());
                }
            });
}
 

@Override
public void start() {
    try {
        gatewayStartLatch = new CountDownLatch(1);

        vertx = Vertx.vertx(new VertxOptions()
                .setBlockedThreadCheckInterval(99999));
        echoServer.start();

        DeploymentOptions options = new DeploymentOptions().
                setConfig(vertxConf);

        vertx.deployVerticle(InitVerticle.class.getCanonicalName(),
                options,
                result -> {
                    System.out.println("Deployed init verticle! " + (result.failed() ? "failed" : "succeeded"));
                    gatewayStartLatch.countDown();
                });

        gatewayStartLatch.await();
    } catch (Exception e) {
        throw new RuntimeException(e);
    }
}
 

@Test
public void connectWithVertxInstance(TestContext context) {
  assumeTrue(options.isUsingDomainSocket());
  Vertx vertx = Vertx.vertx(new VertxOptions().setPreferNativeTransport(true));
  try {
    client = PgPool.pool(vertx, new PgConnectOptions(options), new PoolOptions());
    Async async = context.async();
    client.getConnection(context.asyncAssertSuccess(pgConnection -> {
      async.complete();
      pgConnection.close();
    }));
    async.await();
  } finally {
    vertx.close();
  }
}
 

public static Vertx getVertx(Map<String, String> config) {
    if (vertx == null) {
        synchronized (VertxNetUtils.class) {
            if (vertx == null) {
                Integer blockedThreadCheckInterval = Integer.MAX_VALUE;
                if (config.containsKey("blocked_thread_check_interval")) {
                    blockedThreadCheckInterval = Integer.parseInt(config.get("blocked_thread_check_interval"));
                    if (blockedThreadCheckInterval <= 0)
                        blockedThreadCheckInterval = Integer.MAX_VALUE;
                }
                VertxOptions opt = new VertxOptions();
                opt.setBlockedThreadCheckInterval(blockedThreadCheckInterval);
                vertx = Vertx.vertx(opt);
            }
        }
    }
    return vertx;
}
 

protected void startNodes(int numNodes, VertxOptions options) {
    CountDownLatch latch = new CountDownLatch(numNodes);
    vertices = new Vertx[numNodes];
    for (int i = 0; i < numNodes; i++) {
        int index = i;
        clusteredVertx(options.setClusterHost("localhost").setClusterPort(0).setClustered(true)
                .setClusterManager(getClusterManager()), ar -> {
            try {
                if (ar.failed()) {
                    ar.cause().printStackTrace();
                }
                assertTrue("Failed to start node", ar.succeeded());
                vertices[index] = ar.result();
            }
            finally {
                latch.countDown();
            }
        });
    }
    try {
        assertTrue(latch.await(2, TimeUnit.MINUTES));
    } catch (InterruptedException e) {
        fail(e.getMessage());
    }
}
 

@Override
protected void doStart() {
    instances = (instances < 1) ? VertxOptions.DEFAULT_EVENT_LOOP_POOL_SIZE : instances;
    logger.info("Starting Vertx container and deploy Gateway Verticles [{} instance(s)]", instances);

    DeploymentOptions options = new DeploymentOptions().setInstances(instances);

    Single<String> deployment = vertx.rxDeployVerticle(SpringVerticleFactory.VERTICLE_PREFIX + ':' + GraviteeVerticle.class.getName(), options);

    deployment.subscribe(id -> {
        // Deployed
        deploymentId = id;
    }, err -> {
        // Could not deploy
        logger.error("Unable to start HTTP server", err.getCause());

        // HTTP Server is a required component. Shutdown if not available
        Runtime.getRuntime().exit(1);
    });
}
 

public void liteMemberConfig() {
  Config hazelcastConfig = ConfigUtil.loadConfig()
    .setLiteMember(true);

  ClusterManager mgr = new HazelcastClusterManager(hazelcastConfig);

  VertxOptions options = new VertxOptions().setClusterManager(mgr);

  Vertx.clusteredVertx(options, res -> {
    if (res.succeeded()) {
      Vertx vertx = res.result();
    } else {
      // failed!
    }
  });
}
 

/**
 * Configure Dropwizard helper.
 * @param graphiteHost graphite server host
 * @param port  graphits server port
 * @param tu time unit
 * @param period reporting period
 * @param vopt Vert.x options
 * @param hostName logical hostname for this node (reporting)
 */
public static void config(String graphiteHost, int port, TimeUnit tu,
        int period, VertxOptions vopt, String hostName) {
  final String registryName = "okapi";
  MetricRegistry registry = SharedMetricRegistries.getOrCreate(registryName);

  DropwizardMetricsOptions metricsOpt = new DropwizardMetricsOptions();
  metricsOpt.setEnabled(true).setRegistryName(registryName);
  vopt.setMetricsOptions(metricsOpt);
  Graphite graphite = new Graphite(new InetSocketAddress(graphiteHost, port));
  final String prefix = "folio.okapi." + hostName;
  GraphiteReporter reporter = GraphiteReporter.forRegistry(registry)
          .prefixedWith(prefix)
          .build(graphite);
  reporter.start(period, tu);

  logger.info("Metrics remote {}:{} this {}", graphiteHost, port, prefix);
}
 

@Test
public void shouldStartEmbeddedServer(TestContext context) {
  vertx = Vertx.vertx(new VertxOptions()
    .setMetricsOptions(new MicrometerMetricsOptions()
      .setPrometheusOptions(new VertxPrometheusOptions().setEnabled(true)
        .setStartEmbeddedServer(true)
        .setEmbeddedServerOptions(new HttpServerOptions().setPort(9090)))
      .addLabels(Label.LOCAL, Label.HTTP_PATH, Label.REMOTE)
      .setEnabled(true)));

  Async async = context.async();
  // First "blank" connection to trigger some metrics
  PrometheusTestHelper.tryConnect(vertx, context, 9090, "localhost", "/metrics", r1 -> {
    // Delay to make "sure" metrics are populated
    vertx.setTimer(500, l ->
      // Second connection, this time actually reading the metrics content
      PrometheusTestHelper.tryConnect(vertx, context, 9090, "localhost", "/metrics", body -> {
          context.verify(v2 -> assertThat(body.toString())
            .contains("vertx_http_client_requests{local=\"?\",method=\"GET\",path=\"/metrics\",remote=\"localhost:9090\"")
            .doesNotContain("vertx_http_client_responseTime_seconds_bucket"));
          async.complete();
      }));
  });
  async.awaitSuccess(10000);
}
 

@Test
public void shouldBindExistingServer(TestContext context) {
  vertx = Vertx.vertx(new VertxOptions()
    .setMetricsOptions(new MicrometerMetricsOptions()
      .setPrometheusOptions(new VertxPrometheusOptions().setEnabled(true))
      .setEnabled(true)));

  Router router = Router.router(vertx);
  router.route("/custom").handler(routingContext -> {
    PrometheusMeterRegistry prometheusRegistry = (PrometheusMeterRegistry) BackendRegistries.getDefaultNow();
    String response = prometheusRegistry.scrape();
    routingContext.response().end(response);
  });
  vertx.createHttpServer().requestHandler(router).exceptionHandler(context.exceptionHandler()).listen(8081);

  Async async = context.async();
  PrometheusTestHelper.tryConnect(vertx, context, 8081, "localhost", "/custom", body -> {
    context.verify(v -> assertThat(body.toString())
          .contains("vertx_http_"));
    async.complete();
  });
  async.awaitSuccess(10000);
}
 

@Test
public void shouldExcludeCategory(TestContext context) {
  vertx = Vertx.vertx(new VertxOptions()
    .setMetricsOptions(new MicrometerMetricsOptions()
      .setPrometheusOptions(new VertxPrometheusOptions().setEnabled(true)
        .setStartEmbeddedServer(true)
        .setEmbeddedServerOptions(new HttpServerOptions().setPort(9090)))
      .addDisabledMetricsCategory(MetricsDomain.HTTP_SERVER)
      .addLabels(Label.LOCAL, Label.REMOTE)
      .setEnabled(true)));

  Async async = context.async();
  PrometheusTestHelper.tryConnect(vertx, context, 9090, "localhost", "/metrics", body -> {
    context.verify(v -> assertThat(body.toString())
      .contains("vertx_http_client_connections{local=\"?\",remote=\"localhost:9090\",} 1.0")
      .doesNotContain("vertx_http_server_connections{local=\"0.0.0.0:9090\",remote=\"_\",} 1.0"));
    async.complete();
  });
  async.awaitSuccess(10000);
}
 

@Test
public void shouldPublishQuantilesWithProvidedRegistry(TestContext context) throws Exception {
  PrometheusMeterRegistry registry = new PrometheusMeterRegistry(PrometheusConfig.DEFAULT);
  vertx = Vertx.vertx(new VertxOptions()
    .setMetricsOptions(new MicrometerMetricsOptions()
      .setPrometheusOptions(new VertxPrometheusOptions().setEnabled(true)
        .setPublishQuantiles(true)
        .setStartEmbeddedServer(true)
        .setEmbeddedServerOptions(new HttpServerOptions().setPort(9090)))
      .setMicrometerRegistry(registry)
      .setEnabled(true)));

  Async async = context.async();
  // Dummy connection to trigger some metrics
  PrometheusTestHelper.tryConnect(vertx, context, 9090, "localhost", "/metrics", r1 -> {
    // Delay to make "sure" metrics are populated
    vertx.setTimer(500, l -> {
      assertThat(registry.scrape()).contains("vertx_http_client_responseTime_seconds_bucket{code=\"200\"");
      async.complete();
    });
  });
  async.awaitSuccess(10000);
}
 

public static void main(String args[]){

        VertxOptions options = new VertxOptions();
        options.setBlockedThreadCheckInterval(20000);
        options.setMaxEventLoopExecuteTime(20000);
        SummerServer summerServer =SummerServer.create("localhost",8080,options);

        DeploymentOptions deploymentOptions = new DeploymentOptions();
        summerServer.getVertx().
                deployVerticle(MyVerticle.class.getName());
        deploymentOptions.setWorker(true);
        summerServer.getSummerRouter().registerResource(Hello.class);
        summerServer.getVertx().
                deployVerticle(SummerServer.WebServer.class.getName());
        summerServer.start(deploymentOptions);
    }
 

/**
 * 设置vert.x配置
 */
@Override
public void beforeStartingVertx(VertxOptions options) {
	try {
		byte[] bytes = Files.readAllBytes(PathUtil.getPath("conf.json"));
		Buffer buff = Buffer.buffer(bytes);
		// 总配置文件
		JsonObject conf = buff.toJsonObject();
		// vert.x配置文件
		JsonObject vertxc = conf.getJsonObject("vertx", getDefaultVertxConfig());
		initVertxConfig(vertxc, options);
		// 集群配置文件
		JsonObject clusterc = conf.getJsonObject("cluster", new JsonObject().put("clusterType", CLUSTER_TYPE));
		if (!CLUSTER_TYPE.equals(clusterc.getString("clusterType"))) {
			ClusterManager cmgr = VxApiClusterManagerFactory.getClusterManager(clusterc.getString("clusterType"),
					clusterc.getJsonObject("clusterConf", getDefaultClusterConfig()));
			options.setClusterManager(cmgr);
			options.setClustered(true);
		}
	} catch (IOException e) {
		throw new FileSystemException(e);
	}
}
 

@Before
public void setUp() {
  VertxOptions options = new VertxOptions()
    .setClusterManager(new IgniteClusterManager());
  Vertx.clusteredVertx(options, ar -> {
    vertx = ar.result();
  });
  await().until(() -> vertx != null);
  discovery = new DiscoveryImpl(vertx, new ServiceDiscoveryOptions());
}
 

@BeforeAll
public static void before() {
    vertx = Vertx.vertx(new VertxOptions().setMetricsOptions(
            new MicrometerMetricsOptions()
                    .setPrometheusOptions(new VertxPrometheusOptions().setEnabled(true))
                    .setEnabled(true)
    ));
}
 

@Before
public void setup() {
  vertx = Vertx.vertx(new VertxOptions().setMetricsOptions(
    new MetricsOptions().setEnabled(true).setFactory(tracingOptions -> new VertxMetrics() {
      @Override
      public ClientMetrics<?, ?, ?, ?> createClientMetrics(SocketAddress remoteAddress, String type, String namespace) {
        return metrics;
      }
    }))
  );
}
 

public static void run(DeploymentOptions options, Class clazz) {
    VertxOptions vOpts = new VertxOptions();
    vOpts.setClustered(true);
    Vertx.clusteredVertx(vOpts, cluster -> {
        if (cluster.succeeded()) {
            final Vertx result = cluster.result();
            result.deployVerticle(clazz.getName(), options, handle -> {

            });
        }
    });
}
 

@Override
public void configure(JsonNode nodeConfig) {
    helper = new Vertx3GatewayHelper();
    vertxConf = helper.loadJsonObjectFromResources(nodeConfig, "config");
    apiToFilePushEmulatorConfig = helper.loadJsonObjectFromResources(nodeConfig, "configPushEmulator");

    API_PORT = helper.getApiPortDynamically(apiToFilePushEmulatorConfig);
    GW_PORT = helper.getGatewayPortDynamically(apiToFilePushEmulatorConfig);

    apiToFileVx = Vertx.vertx(new VertxOptions()
            .setBlockedThreadCheckInterval(99999));

    apiToFileVx.deployVerticle(ApiVerticle.class.getCanonicalName(),
            new DeploymentOptions().setConfig(apiToFilePushEmulatorConfig),
            complete -> a2fInitLatch.countDown());

    apiToFileVx.eventBus().consumer("reset").handler(reset -> resetLatch.countDown());
    apiToFileVx.eventBus().consumer("rewrite").handler(rewritten -> rewriteCdl.countDown());

    // Important: before we deploy the verticle, we must ensure that the API-to-File registry
    // has fully initialised. Otherwise the variable may not be substituted yet.
    try {
        a2fInitLatch.await();
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
}
 

@BeforeAll
public static void before() {
    VertxOptions options = new VertxOptions().setMetricsOptions(
            new MicrometerMetricsOptions()
                    .setPrometheusOptions(new VertxPrometheusOptions().setEnabled(true))
                    .setEnabled(true));
    vertx = Vertx.vertx(options);
}
 

@Before
public void startVertxWithConfiguredVertxOptions(TestContext context) throws IOException {
    Async async = context.async();

    server = new Server() {
        @Override
        protected @NonNull
        VertxOptions configureVertxOptions(VertxOptions options) {
            return options.setWorkerPoolSize(3);
        }

        @Override
        protected AuthProvider setupAuthenticationRoutes() {
            Router router = AppGlobals.get().getRouter();
            router.get("/test").handler(context -> {
                context.response().end("OK");
            });
            return super.setupAuthenticationRoutes();
        }
    };
    server.start(new JsonObject().put("sessionDisabled", true), TestResource.class, TestResourceRxJava1.class)
            .subscribe(() -> {
                webClient = WebClient.create(server.getVertx(),
                        new WebClientOptions().setDefaultHost("localhost").setDefaultPort(9000));
                async.complete();
            }, x -> {
                x.printStackTrace();
                context.fail(x);
                async.complete();
            });
}
 

public void example2() {
  IgniteConfiguration cfg = new IgniteConfiguration();
  // Configuration code (omitted)

  ClusterManager clusterManager = new IgniteClusterManager(cfg);

  VertxOptions options = new VertxOptions().setClusterManager(clusterManager);
  Vertx.clusteredVertx(options, res -> {
    if (res.succeeded()) {
      Vertx vertx = res.result();
    } else {
      // failed!
    }
  });
}
 

public void setupMonitoredUrisWithAliases() {
  Vertx vertx = Vertx.vertx(new VertxOptions().setMetricsOptions(
      new DropwizardMetricsOptions().
          setEnabled(true).
          addMonitoredHttpServerUri(new Match().setValue("/users/.*").setAlias("users").setType(MatchType.REGEX))
  ));
}
 

public void example1() {

    ClusterManager mgr = new HazelcastClusterManager();

    VertxOptions options = new VertxOptions().setClusterManager(mgr);

    Vertx.clusteredVertx(options, res -> {
      if (res.succeeded()) {
        Vertx vertx = res.result();
      } else {
        // failed!
      }
    });
  }
 

public static Vertx create() {
    VertxOptions options = new VertxOptions()
            .setWorkerPoolSize(1)
            .setInternalBlockingPoolSize(1)
            .setEventLoopPoolSize(1);
    return Vertx.vertx(options);
}
 

@BeforeClass
public static void setUp(TestContext context) throws Exception {
  Async async = context.async();
  Vertx.clusteredVertx(new VertxOptions(), r -> {
    if (r.succeeded()) {
      Vertx vertx = r.result();
      kue = Kue.createQueue(vertx, new JsonObject());
      vertx.deployVerticle(new KueVerticle(), r2 -> {
        if (r2.succeeded()) {
          kue.jobRangeByType(TYPE, "inactive", 0, 100, "asc").setHandler(r1 -> {
            if (r1.succeeded()) {
              r1.result().forEach(Job::remove);
              vertx.deployVerticle(new KueHttpVerticle(), r3 -> {
                if (r3.succeeded())
                  async.complete();
                else
                  context.fail(r3.cause());
              });
            } else {
              context.fail(r1.cause());
            }
          });
        } else {
          context.fail(r2.cause());
        }
      });

    } else {
      context.fail(r.cause());
    }
  });
}
 

@Test
public void testFromJson() throws Exception {
  VertxMetricsFactoryImpl vmfi = new VertxMetricsFactoryImpl();
  VertxMetricsImpl metrics = (VertxMetricsImpl) vmfi.metrics(new VertxOptions(
    new JsonObject().put("metricsOptions", new JsonObject()
      .put("enabled", true)
      .put("monitoredEventBusHandlers", new JsonArray()
        .add(new JsonObject().put("value", "foo")))
      .put("monitoredHttpServerUris", new JsonArray()
        .add(new JsonObject().put("value", "http://www.bar.com")))
      .put("monitoredHttpClientUris", new JsonArray()
        .add(new JsonObject().put("value", "http://www.baz.com")))
      .put("monitoredHttpClientEndpoints", new JsonArray()
        .add(new JsonObject().put("value", "http://www.foobar.com"))))
  ));
  DropwizardMetricsOptions options = metrics.getOptions();

  assertEquals(1, options.getMonitoredEventBusHandlers().size());
  assertEquals("foo", options.getMonitoredEventBusHandlers().get(0).getValue());
  assertEquals(MatchType.EQUALS, options.getMonitoredEventBusHandlers().get(0).getType());

  assertEquals(1, options.getMonitoredHttpServerUris().size());
  assertEquals("http://www.bar.com", options.getMonitoredHttpServerUris().get(0).getValue());
  assertEquals(MatchType.EQUALS, options.getMonitoredHttpServerUris().get(0).getType());

  assertEquals(1, options.getMonitoredHttpClientUris().size());
  assertEquals("http://www.baz.com", options.getMonitoredHttpClientUris().get(0).getValue());
  assertEquals(MatchType.EQUALS, options.getMonitoredHttpClientUris().get(0).getType());

  assertEquals(1, options.getMonitoredHttpClientEndpoint().size());
  assertEquals("http://www.foobar.com", options.getMonitoredHttpClientEndpoint().get(0).getValue());
  assertEquals(MatchType.EQUALS, options.getMonitoredHttpClientEndpoint().get(0).getType());
}
 

@BeforeAll
public static void before() {
    //Setup Micrometer metrics options
    VertxOptions options = new VertxOptions().setMetricsOptions(
            new MicrometerMetricsOptions()
                    .setPrometheusOptions(new VertxPrometheusOptions().setEnabled(true))
                    .setEnabled(true));
    vertx = Vertx.vertx(options);
}