万字长文入门Vert.x

 1<dependency>
 2    <groupId>io.vertx</groupId>
 3    <artifactId>vertx-web</artifactId>
 4    <version>${vertx.version}</version>
 5  </dependency>
 6  
 7<dependency>
 8    <groupId>io.vertx</groupId>
 9    <artifactId>vertx-unit</artifactId>
10    <version>${vertx.version}</version>
11    <scope>test</scope>
12</dependency>

 1HttpServer server = vertx.createHttpServer();
 2
 3server.requestHandler(request -> {
 4
 5  // This handler gets called for each request that arrives on the server
 6  HttpServerResponse response = request.response();
 7  response.putHeader("content-type", "text/plain");
 8
 9  // Write to the response and end it
10  response.end("Hello World!");
11});
12
13server.listen(8080);

 1Router router = Router.router(vertx);
 2router
 3        .get("/api/health-check")
 4        .handler(
 5            context ->
 6                context
 7                    .response()
 8                    .setStatusCode(200)
 9                    .putHeader("content-type", "application/json;charset=utf-8")
10                    .putHeader("cache-control", "no-cache")
11                    .end(
12                        new JsonObject()
13                            .put("code", 0)
14                            .put("time", System.currentTimeMillis())
15                            .toBuffer()));

1vertx.executeBlocking(promise -> {
2  // Call some blocking API that takes a significant amount of time to return
3  String result = someAPI.blockingMethod("hello");
4  promise.complete(result);
5}, res -> {
6  System.out.println("The result is: " + res.result());
7});

 1 /**
 2   * Safely execute some blocking code.
 3   * <p>
 4   * Executes the blocking code in the handler {@code blockingCodeHandler} using a thread from the worker pool.
 5   * <p>
 6   * When the code is complete the handler {@code resultHandler} will be called with the result on the original context
 7   * (e.g. on the original event loop of the caller).
 8   * <p>
 9   * A {@code Future} instance is passed into {@code blockingCodeHandler}. When the blocking code successfully completes,
10   * the handler should call the {@link Promise#complete} or {@link Promise#complete(Object)} method, or the {@link Promise#fail}
11   * method if it failed.
12   * <p>
13   * In the {@code blockingCodeHandler} the current context remains the original context and therefore any task
14   * scheduled in the {@code blockingCodeHandler} will be executed on the this context and not on the worker thread.
15   * <p>
16   * The blocking code should block for a reasonable amount of time (i.e no more than a few seconds). Long blocking operations
17   * or polling operations (i.e a thread that spin in a loop polling events in a blocking fashion) are precluded.
18   * <p>
19   * When the blocking operation lasts more than the 10 seconds, a message will be printed on the console by the
20   * blocked thread checker.
21   * <p>
22   * Long blocking operations should use a dedicated thread managed by the application, which can interact with
23   * verticles using the event-bus or {@link Context#runOnContext(Handler)}
24   *
25   * @param blockingCodeHandler  handler representing the blocking code to run
26   * @param resultHandler  handler that will be called when the blocking code is complete
27   * @param ordered  if true then if executeBlocking is called several times on the same context, the executions
28   *                 for that context will be executed serially, not in parallel. if false then they will be no ordering
29   *                 guarantees
30   * @param <T> the type of the result
31   */
32  <T> void executeBlocking(Handler<Promise<T>> blockingCodeHandler, boolean ordered, Handler<AsyncResult<@Nullable T>> resultHandler);

1  public <T> void executeBlocking(Handler<Promise<T>> blockingCodeHandler, boolean ordered,
2                                  Handler<AsyncResult<T>> asyncResultHandler) {
3    ContextInternal context = getOrCreateContext();
4    context.executeBlocking(blockingCodeHandler, ordered, asyncResultHandler);
5  }

 1public ContextInternal getOrCreateContext() {
 2    ContextInternal ctx = getContext();
 3    if (ctx == null) {
 4      // We are running embedded - Create a context
 5      ctx = createEventLoopContext(null, null, null, Thread.currentThread().getContextClassLoader());
 6    }
 7    return ctx;
 8  }
 9
10 public AbstractContext getContext() {
11    AbstractContext context = (AbstractContext) ContextInternal.current();
12    if (context != null && context.owner() == this) {
13      return context;
14    }
15    return null;
16  }
17
18static ContextInternal current() {
19    Thread current = Thread.currentThread();
20    if (current instanceof VertxThread) {
21      return ((VertxThread) current).context();
22    } else if (current instanceof FastThreadLocalThread) {
23      return AbstractContext.holderLocal.get().ctx;
24    }
25    return null;
26  }

 1public final class VertxThread extends FastThreadLocalThread implements BlockedThreadChecker.Task {
 2
 3  static final String DISABLE_TCCL_PROP_NAME = "vertx.disableTCCL";
 4  static final boolean DISABLE_TCCL = Boolean.getBoolean(DISABLE_TCCL_PROP_NAME);
 5
 6  private final boolean worker;
 7  private final long maxExecTime;
 8  private final TimeUnit maxExecTimeUnit;
 9  private long execStart;
10  private ContextInternal context;
11
12  public VertxThread(Runnable target, String name, boolean worker, long maxExecTime, TimeUnit maxExecTimeUnit) {
13    super(target, name);
14    this.worker = worker;
15    this.maxExecTime = maxExecTime;
16    this.maxExecTimeUnit = maxExecTimeUnit;
17  }
18
19  /**
20   * @return the current context of this thread, this method must be called from the current thread
21   */
22  ContextInternal context() {
23    return context;
24  }
25
26  private void executeStart() {
27    if (context == null) {
28      execStart = System.nanoTime();
29    }
30  }
31
32  private void executeEnd() {
33    if (context == null) {
34      execStart = 0;
35    }
36  }
37
38  public long startTime() {
39    return execStart;
40  }
41
42  public boolean isWorker() {
43    return worker;
44  }
45
46  @Override
47  public long maxExecTime() {
48    return maxExecTime;
49  }
50
51  @Override
52  public TimeUnit maxExecTimeUnit() {
53    return maxExecTimeUnit;
54  }
55
56  /**
57   * Begin the emission of a context event.
58   * <p>
59   * This is a low level interface that should not be used, instead {@link ContextInternal#emit(Object, io.vertx.core.Handler)}
60   * shall be used.
61   *
62   * @param context the context on which the event is emitted on
63   * @return the current context that shall be restored
64   */
65  ContextInternal beginEmission(ContextInternal context) {
66    if (!ContextImpl.DISABLE_TIMINGS) {
67      executeStart();
68    }
69    ContextInternal prev = this.context;
70    this.context = context;
71    return prev;
72  }
73
74  /**
75   * End the emission of a context task.
76   * <p>
77   * This is a low level interface that should not be used, instead {@link ContextInternal#emit(Object, io.vertx.core.Handler)}
78   * shall be used.
79   *
80   * @param prev the previous context thread to restore, might be {@code null}
81   */
82  void endEmission(ContextInternal prev) {
83    // We don't unset the context after execution - this is done later when the context is closed via
84    // VertxThreadFactory
85    context = prev;
86    if (!ContextImpl.DISABLE_TIMINGS) {
87      executeEnd();
88    }
89  }
90}

 1static <T> Future<T> executeBlocking(ContextInternal context, Handler<Promise<T>> blockingCodeHandler,
 2      WorkerPool workerPool, TaskQueue queue) {
 3    PoolMetrics metrics = workerPool.metrics();
 4    Object queueMetric = metrics != null ? metrics.submitted() : null;
 5    Promise<T> promise = context.promise();
 6    Future<T> fut = promise.future();
 7    try {
 8      Runnable command = () -> {
 9        Object execMetric = null;
10        if (metrics != null) {
11          execMetric = metrics.begin(queueMetric);
12        }
13        context.emit(promise, f -> {
14          try {
15            blockingCodeHandler.handle(promise);
16          } catch (Throwable e) {
17            promise.tryFail(e);
18          }
19        });
20        if (metrics != null) {
21          metrics.end(execMetric, fut.succeeded());
22        }
23      };
24      Executor exec = workerPool.executor();
25      if (queue != null) {
26        queue.execute(command, exec);
27      } else {
28        exec.execute(command);
29      }
30    } catch (RejectedExecutionException e) {
31      // Pool is already shut down
32      if (metrics != null) {
33        metrics.rejected(queueMetric);
34      }
35      throw e;
36    }
37    return fut;
38  }

 1public final class VertxContextExecutor implements Executor {
 2
 3  private final Vertx vertx;
 4
 5  public VertxContextExecutor(@NonNull Vertx vertx) {
 6    this.vertx = vertx;
 7  }
 8
 9  @Override
10  public void execute(@NonNull Runnable command) {
11    vertx.runOnContext(v -> command.run());
12  }
13}

 1Caffeine.newBuilder()
 2            .maximumSize(200)
 3            .executor(new VertxContextExecutor(vertx))
 4            .refreshAfterWrite(Duration.ofMinutes(2))
 5            .buildAsync(
 6                new AsyncCacheLoader<>() {
 7                  @Override
 8                  public @NonNull CompletableFuture<Optional<HashFunction>> asyncLoad(
 9                      @NonNull AuthEntry key, @NonNull Executor executor) {
10                    return vertx
11                        .executeBlocking(
12                            (Promise<Optional<HashFunction>> promise) ->
13                                loadHashFunction(key.getAppId(), promise),
14                            false)
15                        .toCompletionStage()
16                        .toCompletableFuture();
17                  }
18                });

 1import com.google.common.base.Preconditions;
 2import io.vertx.core.Vertx;
 3import lombok.NonNull;
 4import lombok.extern.slf4j.Slf4j;
 5import net.jodah.failsafe.util.concurrent.DefaultScheduledFuture;
 6import net.jodah.failsafe.util.concurrent.Scheduler;
 7
 8import java.util.concurrent.Callable;
 9import java.util.concurrent.ScheduledFuture;
10import java.util.concurrent.TimeUnit;
11
12/**
13 * @author fishzhao
14 * @since 2021-01-07
15 */
16@Slf4j
17public final class FailsafeVertxScheduler implements Scheduler {
18
19  private final Vertx vertx;
20
21  public FailsafeVertxScheduler(@NonNull Vertx vertx) {
22    this.vertx = vertx;
23  }
24
25  private static void execute(Callable<?> callable) {
26    try {
27      callable.call();
28    } catch (Exception e) {
29      log.error("Failsafe threw exception: ", e);
30    }
31  }
32
33  @Override
34  public ScheduledFuture<?> schedule(Callable<?> callable, long l, TimeUnit timeUnit) {
35    Preconditions.checkArgument(l >= 0, "illegal delay: %s", l);
36    if (l == 0) {
37      vertx.runOnContext(v -> execute(callable));
38      return new VertxScheduledFuture<>(null, -1);
39    }
40    long timerId = vertx.setTimer(timeUnit.toMillis(l), id -> execute(callable));
41    return new VertxScheduledFuture<>(vertx, timerId);
42  }
43
44  private static final class VertxScheduledFuture<T> extends DefaultScheduledFuture<T> {
45    private final Vertx vertx;
46    private final long timerId;
47
48    public VertxScheduledFuture(Vertx vertx, long timerId) {
49      this.vertx = vertx;
50      this.timerId = timerId;
51    }
52
53    @Override
54    public boolean cancel(boolean mayInterruptIfRunning) {
55      if (timerId >= 0 && vertx != null) {
56        return vertx.cancelTimer(timerId);
57      }
58      return false;
59    }
60  }
61}

1private Future<SendResult> sendBytesRetry(String tid, List<byte[]> bytes) {
2    // Benchmark                 Mode  Cnt    Score    Error  Units
3    // UUIDPerformanceTest.uuid  avgt   25  684.833 ± 22.660  ns/op
4    String uuid = UUID.randomUUID().toString();
5    CompletableFuture<SendResult> completableFuture =
6        Failsafe.with(retryPolicy).with(scheduler).getStageAsync(() -> sendBytes(tid, uuid, bytes));
7    return Future.fromCompletionStage(completableFuture);
8  }

1        <dependency>
2            <groupId>io.netty</groupId>
3            <artifactId>netty-transport-native-epoll</artifactId>
4            <version>${netty.version}</version>
5            <classifier>linux-x86_64</classifier>
6        </dependency>

1Vertx vertx = Vertx.vertx(new VertxOptions().setPreferNativeTransport(true));
2vertx.createHttpServer(
3            new HttpServerOptions().setTcpNoDelay(true).setTcpFastOpen(true).setTcpQuickAck(true));