import io.vertx.circuitbreaker.CircuitBreaker; //导入依赖的package包/类
public void example2(Vertx vertx) {
  CircuitBreaker breaker = CircuitBreaker.create("my-circuit-breaker", vertx,
      new CircuitBreakerOptions().setMaxFailures(5).setTimeout(2000)
  );

  breaker.<String>execute(future -> {
    vertx.createHttpClient().getNow(8080, "localhost", "/", response -> {
      if (response.statusCode() != 200) {
        future.fail("HTTP error");
      } else {
        response
            .exceptionHandler(future::fail)
            .bodyHandler(buffer -> {
              future.complete(buffer.toString());
            });
      }
    });
  }).setHandler(ar -> {
    // Do something with the result
  });
}
 

import io.vertx.circuitbreaker.CircuitBreaker; //导入依赖的package包/类
private Router buildRouter(DfxConfig config, Map<String, Accessible> accessibleMap) {
    Router router = Router.router(vertx);

    try {
        addCorsHandler(router, config.getCors());
    }catch (Exception e) {
        logger.error("Adding CORS handler failed, cause: {}", e);
        vertx.close();
        System.exit(-1);
    }

    config.getMappings().forEach((key, value) -> {
                CircuitBreaker circuitBreaker = CircuitBreaker.create(key, vertx
                        , config.getCircuitBreakerOptions());
                if (accessibleMap.get(value) != null) {
                    router.route(key).handler(new AccessibleHandler(key, accessibleMap.get(value)
                            , circuitBreaker, vertx));
                }
            }
    );

    return router;
}
 

import io.vertx.circuitbreaker.CircuitBreaker; //导入依赖的package包/类
public static void withCircuitBreaker(Vertx vertx, CircuitBreaker circuitBreaker
        , Accessible accessible, Map params, Handler<Map> successHandler, Handler<Throwable> failureHandler) {
    circuitBreaker.execute(future ->
            vertx.executeBlocking(f -> {
                        try {
                            f.complete(accessible.invoke(params));
                        } catch (Throwable throwable) {
                            f.fail(throwable);
                        }
                    }
                    , result -> {
                        if (result.succeeded()) {
                            future.complete(result.result());
                        } else {
                            future.fail(result.cause());
                        }
                    })
    ).setHandler(result -> {
        if (result.succeeded()) {
            successHandler.handle((Map) result.result());
        } else {
            logger.error("CB[{}] execution failed, cause: ", circuitBreaker.name(), result.cause());
            failureHandler.handle(result.cause());
        }
    });
}
 

import io.vertx.circuitbreaker.CircuitBreaker; //导入依赖的package包/类
public void example1(Vertx vertx) {
  CircuitBreaker breaker = CircuitBreaker.create("my-circuit-breaker", vertx,
      new CircuitBreakerOptions()
          .setMaxFailures(5) // number of failure before opening the circuit
          .setTimeout(2000) // consider a failure if the operation does not succeed in time
          .setFallbackOnFailure(true) // do we call the fallback on failure
          .setResetTimeout(10000) // time spent in open state before attempting to re-try
  );

  breaker.execute(future -> {
    // some code executing with the breaker
    // the code reports failures or success on the given future.
    // if this future is marked as failed, the breaker increased the
    // number of failures
  }).setHandler(ar -> {
    // Get the operation result.
  });
}
 

import io.vertx.circuitbreaker.CircuitBreaker; //导入依赖的package包/类
public void example3(Vertx vertx) {
  CircuitBreaker breaker = CircuitBreaker.create("my-circuit-breaker", vertx,
      new CircuitBreakerOptions().setMaxFailures(5).setTimeout(2000)
  );

  breaker.executeWithFallback(
      future -> {
        vertx.createHttpClient().getNow(8080, "localhost", "/", response -> {
          if (response.statusCode() != 200) {
            future.fail("HTTP error");
          } else {
            response
                .exceptionHandler(future::fail)
                .bodyHandler(buffer -> {
                  future.complete(buffer.toString());
                });
          }
        });
      }, v -> {
        // Executed when the circuit is opened
        return "Hello";
      })
      .setHandler(ar -> {
        // Do something with the result
      });
}
 

import io.vertx.circuitbreaker.CircuitBreaker; //导入依赖的package包/类
public void example4(Vertx vertx) {
  CircuitBreaker breaker = CircuitBreaker.create("my-circuit-breaker", vertx,
      new CircuitBreakerOptions().setMaxFailures(5).setTimeout(2000)
  ).fallback(v -> {
    // Executed when the circuit is opened.
    return "hello";
  });

  breaker.execute(
      future -> {
        vertx.createHttpClient().getNow(8080, "localhost", "/", response -> {
          if (response.statusCode() != 200) {
            future.fail("HTTP error");
          } else {
            response
                .exceptionHandler(future::fail)
                .bodyHandler(buffer -> {
                  future.complete(buffer.toString());
                });
          }
        });
      });
}
 

import io.vertx.circuitbreaker.CircuitBreaker; //导入依赖的package包/类
public void example5(Vertx vertx) {
  CircuitBreaker breaker = CircuitBreaker.create("my-circuit-breaker", vertx,
      new CircuitBreakerOptions().setMaxFailures(5).setTimeout(2000)
  ).openHandler(v -> {
    System.out.println("Circuit opened");
  }).closeHandler(v -> {
    System.out.println("Circuit closed");
  });

  breaker.execute(
      future -> {
        vertx.createHttpClient().getNow(8080, "localhost", "/", response -> {
          if (response.statusCode() != 200) {
            future.fail("HTTP error");
          } else {
            // Do something with the response
            future.complete();
          }
        });
      });
}
 

import io.vertx.circuitbreaker.CircuitBreaker; //导入依赖的package包/类
public void example7(Vertx vertx) {
  // Create the circuit breaker as usual.
  CircuitBreaker breaker = CircuitBreaker.create("my-circuit-breaker", vertx);
  CircuitBreaker breaker2 = CircuitBreaker.create("my-second-circuit-breaker", vertx);

  // Create a Vert.x Web router
  Router router = Router.router(vertx);
  // Register the metric handler
  router.get("/hystrix-metrics").handler(HystrixMetricHandler.create(vertx));

  // Create the HTTP server using the router to dispatch the requests
  vertx.createHttpServer()
    .requestHandler(router::accept)
    .listen(8080);

}
 

import io.vertx.circuitbreaker.CircuitBreaker; //导入依赖的package包/类
/**
 * A version of reset that can force the the state to `close` even if the circuit breaker is open. This is an
 * internal API.
 *
 * @param force whether or not we force the state and allow an illegal transition
 * @return the current circuit breaker.
 */
public synchronized CircuitBreaker reset(boolean force) {
  failures = 0;

  if (state == CircuitBreakerState.CLOSED) {
    // Do nothing else.
    return this;
  }

  if (!force && state == CircuitBreakerState.OPEN) {
    // Resetting the circuit breaker while we are in the open state is an illegal transition
    return this;
  }

  state = CircuitBreakerState.CLOSED;
  closeHandler.handle(null);
  sendUpdateOnEventBus();
  return this;
}
 

import io.vertx.circuitbreaker.CircuitBreaker; //导入依赖的package包/类
public static void main(String[] args) {
  Vertx vertx = Vertx.vertx();
  CircuitBreakerOptions options = new CircuitBreakerOptions()
    .setFallbackOnFailure(true)
    .setMaxFailures(10)
    .setResetTimeout(5000)
    .setTimeout(1000)
    .setMetricsRollingWindow(10000);

  CircuitBreaker cba = CircuitBreaker.create("A", vertx, options);
  CircuitBreaker cbb = CircuitBreaker.create("B", vertx, options);
  CircuitBreaker cbc = CircuitBreaker.create("C", vertx, options);

  Router router = Router.router(vertx);
  router.get("/metrics").handler(HystrixMetricHandler.create(vertx));


  router.get("/A").handler(rc -> a(rc, cba));
  router.get("/B").handler(rc -> b(rc, cbb));
  router.get("/C").handler(rc -> c(rc, cbc));

  vertx.createHttpServer()
    .requestHandler(router::accept)
    .listen(8080);
}
 

import io.vertx.circuitbreaker.CircuitBreaker; //导入依赖的package包/类
private static void b(RoutingContext rc, CircuitBreaker cb) {
  int choice = random.nextInt(10);
  if (choice < 5) {
    cb.executeWithFallback(
      commandThatWorks(rc.vertx()),
      (t) -> "OK (fallback)")
      .setHandler(s -> rc.response().end(s.result()));
  } else if (choice < 7) {
    cb.executeWithFallback(
      commandThatCrashes(rc.vertx()),
      (t) -> "OK (fallback)")
      .setHandler(s -> rc.response().end(s.result()));
  } else {
    cb.executeWithFallback(
      commandThatFails(rc.vertx()),
      (t) -> "OK (fallback)")
      .setHandler(s -> rc.response().end(s.result()));
  }
}
 

import io.vertx.circuitbreaker.CircuitBreaker; //导入依赖的package包/类
private static void c(RoutingContext rc, CircuitBreaker cb) {
  int choice = random.nextInt(10);
  if (choice < 5) {
    cb.executeWithFallback(
      commandThatWorks(rc.vertx()),
      (t) -> "OK (fallback)")
      .setHandler(s -> rc.response().end(s.result()));
  } else if (choice < 7) {
    cb.executeWithFallback(
      commandThatTimeout(rc.vertx(), 15000),
      (t) -> "OK (fallback)")
      .setHandler(s -> rc.response().end(s.result()));
  } else {
    cb.executeWithFallback(
      commandThatFails(rc.vertx()),
      (t) -> "OK (fallback)")
      .setHandler(s -> rc.response().end(s.result()));
  }
}
 

import io.vertx.circuitbreaker.CircuitBreaker; //导入依赖的package包/类
@Test
@Repeat(5)
public void testOk() {
  breaker = CircuitBreaker.create("test", vertx, new CircuitBreakerOptions());
  assertThat(breaker.state()).isEqualTo(CircuitBreakerState.CLOSED);

  AtomicBoolean operationCalled = new AtomicBoolean();
  AtomicReference<String> completionCalled = new AtomicReference<>();
  breaker.<String>execute(fut -> {
    operationCalled.set(true);
    fut.complete("hello");
  }).setHandler(ar -> completionCalled.set(ar.result()));

  await().until(operationCalled::get);
  await().until(() -> completionCalled.get().equalsIgnoreCase("hello"));
}
 

import io.vertx.circuitbreaker.CircuitBreaker; //导入依赖的package包/类
@Test
@Repeat(5)
public void testWithUserFutureOk() {
  breaker = CircuitBreaker.create("test", vertx, new CircuitBreakerOptions());
  assertThat(breaker.state()).isEqualTo(CircuitBreakerState.CLOSED);

  AtomicBoolean operationCalled = new AtomicBoolean();
  AtomicReference<String> completionCalled = new AtomicReference<>();

  Future<String> userFuture = Future.future();
  userFuture.setHandler(ar ->
    completionCalled.set(ar.result()));

  breaker.executeAndReport(userFuture, fut -> {
    operationCalled.set(true);
    fut.complete("hello");
  });

  await().until(operationCalled::get);
  await().until(() ->  completionCalled.get().equalsIgnoreCase("hello"));
}
 

import io.vertx.circuitbreaker.CircuitBreaker; //导入依赖的package包/类
@Test
public void testAsynchronousOk() {
  breaker = CircuitBreaker.create("test", vertx, new CircuitBreakerOptions());
  assertThat(breaker.state()).isEqualTo(CircuitBreakerState.CLOSED);

  AtomicBoolean called = new AtomicBoolean();
  AtomicReference<String> result = new AtomicReference<>();
  breaker.<String>execute(future ->
    vertx.setTimer(100, l -> {
      called.set(true);
      future.complete("hello");
    })
  ).setHandler(ar -> result.set(ar.result()));

  await().until(called::get);
  await().untilAtomic(result, is("hello"));
}
 

import io.vertx.circuitbreaker.CircuitBreaker; //导入依赖的package包/类
@Test
public void testAsynchronousWithUserFutureOk() {
  breaker = CircuitBreaker.create("test", vertx, new CircuitBreakerOptions());
  assertThat(breaker.state()).isEqualTo(CircuitBreakerState.CLOSED);

  AtomicBoolean called = new AtomicBoolean();
  AtomicReference<String> result = new AtomicReference<>();

  Future<String> userFuture = Future.future();
  userFuture.setHandler(ar -> result.set(ar.result()));

  breaker.executeAndReport(userFuture, future ->
    vertx.setTimer(100, l -> {
      called.set(true);
      future.complete("hello");
    })
  );

  await().until(called::get);
  await().untilAtomic(result, is("hello"));
}
 

import io.vertx.circuitbreaker.CircuitBreaker; //导入依赖的package包/类
@Before
public void setUp() {
  vertx = Vertx.vertx();
  items.clear();
  cb = CircuitBreaker.create("circuit-breaker", vertx, new CircuitBreakerOptions()
      .setFallbackOnFailure(true)
      .setTimeout(500)
      .setResetTimeout(1000));

  vertx.eventBus().consumer("ok", message -> message.reply("OK"));

  vertx.eventBus().consumer("fail", message -> message.fail(100, "Bad bad bad"));

  vertx.eventBus().consumer("exception", message -> {
    throw new RuntimeException("RT - Bad bad bad");
  });

  vertx.eventBus().consumer("timeout", message -> vertx.setTimer(2000, x -> message.reply("Too late")));
}
 

import io.vertx.circuitbreaker.CircuitBreaker; //导入依赖的package包/类
@Test
public void testWithFailedCommands(TestContext tc) {
  breaker = CircuitBreaker.create("some-circuit-breaker", vertx);
  Async async = tc.async();

  Future<Void> command1 = breaker.execute(commandThatFails());
  Future<Void> command2 = breaker.execute(commandThatWorks());
  Future<Void> command3 = breaker.execute(commandThatWorks());
  Future<Void> command4 = breaker.execute(commandThatFails());

  CompositeFuture.join(command1, command2, command3, command4)
    .setHandler(ar -> {
      assertThat(metrics())
        .contains("name", "some-circuit-breaker")
        .contains("state", CircuitBreakerState.CLOSED.name())
        .contains("totalErrorCount", 2) // Failure + Timeout + Exception
        .contains("totalSuccessCount", 2)
        .contains("totalTimeoutCount", 0)
        .contains("totalExceptionCount", 0)
        .contains("totalFailureCount", 2)
        .contains("totalOperationCount", 4)
        .contains("totalSuccessPercentage", 50)
        .contains("totalErrorPercentage", 50);
      async.complete();
    });
}
 

import io.vertx.circuitbreaker.CircuitBreaker; //导入依赖的package包/类
@Test
public void testEviction(TestContext tc) {
  breaker = CircuitBreaker.create("some-circuit-breaker", vertx,
    new CircuitBreakerOptions().setMetricsRollingWindow(10));
  Async async = tc.async();


  int count = 1000;

  List<Future> list = new ArrayList<>();
  for (int i = 0; i < count; i++) {
    list.add(breaker.execute(commandThatWorks()));
  }

  CompositeFuture.all(list)
    .setHandler(ar -> {
      assertThat(ar).succeeded();
      assertThat(metrics().getInteger("totalOperationCount")).isEqualTo(1000);
      assertThat(metrics().getInteger("rollingOperationCount")).isLessThan(1000);
      async.complete();
    });
}
 

import io.vertx.circuitbreaker.CircuitBreaker; //导入依赖的package包/类
CircuitBreakerRegistryImpl(Vertx vertx, CircuitBreakerRegistryOptions options) {
  this.vertx = vertx;
  this.options = options;
  this.cache = CacheBuilder.newBuilder()
          .expireAfterAccess(options.getCacheExpires(), TimeUnit.SECONDS)
          .removalListener(new RemovalListener<String, CircuitBreaker>() {
            @Override
            public void onRemoval(RemovalNotification<String, CircuitBreaker> notification) {
              Log.create(LOGGER)
                      .setLogType(LogType.LOG)
                      .setModule("CircuitBreaker")
                      .setEvent("cache.removed")
                      .addData("key", notification.getKey())
                      .setMessage("cause by: {}")
                      .addArg(notification.getCause())
                      .info();
            }
          })
          .build(new CacheLoader<String, CircuitBreaker>() {
            @Override
            public CircuitBreaker load(String circuitBreakerName) throws Exception {
              return create(circuitBreakerName);
            }
          });
}