高级 Vert.x 指南

public class MyVerticle extends AbstractVerticle {
  public void start() {
    JsonObject config = context.config();
  }
}

public class Main {
  public static void main(String[] args) {
    WebClient client = WebClient.create(vertx);

    for (int i = 0;i < 4;i++) {
      client
        .get(8080, "myserver.mycompany.com", "/some-uri")
        .send()
        .onSuccess(ar -> {
         // All callbacks are on the same context
        });
    }
  }
}

public class MyVerticle extends AbstractVerticle {
  public void start() {
    Future<HttpServer> future = vertx.createHttpServer()
      .requestHandler(request -> {
        // Executed in the verticle context
      })
      .listen(8080, "localhost");

    future.onComplete(ar -> {
       // Executed in the verticle context
    });
  }
}

Context context = vertx.getOrCreateContext();

Context context = Vertx.currentContext();

public void integrateWithExternalSystem(Handler<Event> handler) {
  // Capture the current context
  Context context = vertx.getOrCreateContext();

  // Run the event handler on the application context
  externalSystem.onEvent(event -> {
    context.runOnContext(v -> handler.handle(event));
  });
}

context.exceptionHandler(throwable -> {
  // Any exception thrown by this context
});

vertx.exceptionHandler(throwable -> {
  // Any exception uncaught exception thrown on context
});

context.reportException(new Exception());

Promise<String> promise = context.promise();

Future<String> future = promise.future();

future.onSuccess(handler);

Promise<String> promise = Promise.promise();

Future<String> future = promise.future();

future.onSuccess(result -> context.emit(result, handler));

Future<String> succeeded = context.succeededFuture("OK usa");
Future<String> failed = context.failedFuture("Oh sorry");

public class MyVerticle extends AbstractVerticle {
  public void start() {
    vertx.createHttpServer()
      .requestHandler(request -> {
        // Executed in a duplicate verticle context
      })
      .listen(8080, "localhost");
  }
}

vertx.createHttpServer()
  .requestHandler(request -> {
    JsonObject specificRequestData = getRequestData(request);
    Context context = vertx.getOrCreateContext();
    context.putLocal("my-stuff", specificRequestData);
    processRequest(request);
  })
  .listen(8080, "localhost");

Context context = vertx.getOrCreateContext();
JsonObject specificRequestData = context.getLocal("my-stuff");

public void startProcessing(Request request) {
  Context duplicate = context.duplicate();
  request.setContext(duplicate);
}

@Override
public void close(Promise<Void> completion) {
  // Do cleanup, the method will complete the future
   doClose(completion);
}

context.addCloseHook(closeable);

context.removeCloseHook(closeable);

public interface TimeServer {

  /**
   * @return a new time server
   */
  static TimeServer create(Vertx vertx) {
    return new TimeServerImpl(vertx);
  }

  /**
   * Set the handler to be called when a time request happens. The handler should complete
   * the future with the time value.
   *
   * @param handler the handler to be called
   * @return this object
   */
  TimeServer requestHandler(Handler<Promise<Long>> handler);

  /**
   * Start and bind the time server.
   *
   * @param port the server port
   * @param host the server host
   * @return the future completed when the socket is bound
   */
  Future<Void> listen(int port, String host);

  /**
   * Close the time server.
   */
  void close();

}

Vertx vertx = Vertx.vertx();

// Create the time server
TimeServer server = TimeServer.create(vertx);
server.requestHandler(time -> {
  time.complete(System.currentTimeMillis());
});

// Start the server
server.listen(8037, "0.0.0.0")
    .onSuccess(v -> System.out.println("Server started"))
    .onFailure(err -> err.printStackTrace());

EventLoopGroup acceptorGroup = vertx.getAcceptorEventLoopGroup(); (1)
EventLoop eventLoop = context.nettyEventLoop(); (2)
bootstrap = new ServerBootstrap(); (3)
bootstrap.channel(NioServerSocketChannel.class);
bootstrap.group(acceptorGroup, eventLoop);
bootstrap.childHandler(new ChannelInitializer<Channel>() {
  @Override
  protected void initChannel(Channel ch) throws Exception {
    ChannelPipeline pipeline = ch.pipeline(); (4)
    TimeServerHandler handler = new TimeServerHandler(context, requestHandler);
    pipeline.addLast(handler);
  }
});

Promise<Void> promise = context.promise(); (1)

ChannelFuture bindFuture = bootstrap.bind(host, port);
bindFuture.addListener(new ChannelFutureListener() {
  @Override
  public void operationComplete(ChannelFuture future) {
    (2)
    if (future.isSuccess()) {
      channel = future.channel();
      promise.complete();
    } else {
      promise.fail(future.cause());
    }
  }
});

return promise.future(); (3)

Promise<Long> result = Promise.promise(); (1)

context.emit(result, requestHandler); (2)

result.future().onComplete(ar -> { (3)
  if (ar.succeeded()) {  (4)
    ByteBuf time = ctx.alloc().buffer(4);
    time.writeInt((int) (ar.result() / 1000L + 2208988800L));
    ChannelFuture f = ctx.writeAndFlush(time);
    f.addListener((ChannelFutureListener) channelFuture -> ctx.close());
  } else {  (5)
    ctx.close();
  }
});

public interface TimeClient {

  /**
   * @return a new time client
   */
  static TimeClient create(Vertx vertx) {
    return new TimeClientImpl(vertx);
  }

  /**
   * Fetch the current time from a server.
   *
   * @param port the server port
   * @param host the server host name
   * @return the result future
   */
  Future<Long> getTime(int port, String host);

}

Vertx vertx = Vertx.vertx();

// Create the time client
TimeClient server = TimeClient.create(vertx);

// Fetch the time
server.getTime(8037, "localhost").onComplete(ar -> {
  if (ar.succeeded()) {
    System.out.println("Time is " + new Date(ar.result()));
  } else {
    ar.cause().printStackTrace();
  }
});

EventLoop eventLoop = context.nettyEventLoop();  (1)

// Create and configure the Netty bootstrap
Bootstrap bootstrap = new Bootstrap(); (2)
bootstrap.group(eventLoop);
bootstrap.channel(NioSocketChannel.class);
bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
bootstrap.handler(new ChannelInitializer<Channel>() {
  @Override
  protected void initChannel(Channel ch) {
    ChannelPipeline pipeline = ch.pipeline(); (3)
    pipeline.addLast(new TimeClientHandler(result));
  }
});

return bootstrap;

ChannelFuture connectFuture = bootstrap.connect(host, port); (1)
connectFuture.addListener(new ChannelFutureListener() {
  @Override
  public void operationComplete(ChannelFuture future) throws Exception {
    if (!future.isSuccess()) {
      result.fail(future.cause()); // 2
    }
  }
});

ByteBuf m = (ByteBuf) msg;
long currentTimeMillis;
try {
  currentTimeMillis = (m.readUnsignedInt() - 2208988800L) * 1000L; (1)
  resultPromise.complete(currentTimeMillis);  (2)
  resultPromise = null; (3)
  ctx.close(); (4)
} finally {
  m.release();
}

Vertx vertx = Vertx.vertx();

MemcachedClient.connect(vertx, 11211, "localhost")
    .compose(client -> {
      System.out.println("connected");

      // Put a value
      return client.set("foo", "bar").compose(v -> {
        System.out.println("Put successful");

        // Now retrieve the same value
        return client.get("foo");
      });
    }).onSuccess(res -> {
      System.out.println("Get successful " + res + "");
    }).onFailure(err -> err.printStackTrace());

> docker run --rm --name my-memcache -p 11211:11211 -d memcached

public interface MemcachedClient {

  /**
   * Connect to memcached, the {@code completionHandler} will get the {@link MemcachedClient} instance.
   */
  static Future<MemcachedClient> connect(Vertx vertx, int port, String host) {
    return MemcachedClientImpl.connect(vertx, port, host, null);
  }

  /**
   * Connect to memcached, the {@code completionHandler} will get the {@link MemcachedClient} instance.
   */
  static Future<MemcachedClient> connect(Vertx vertx, int port, String host, NetClientOptions options) {
    return MemcachedClientImpl.connect(vertx, port, host, options);
  }

  /**
   * Get a cached entry.
   *
   * @param key the entry key
   * @return the result future
   */
  Future<@Nullable String> get(String key);

  /**
   * Set a cached entry.
   *
   * @param key the entry key
   * @param value the entry value
   * @return the result future
   */
  Future<Void> set(String key, String value);

}

NetClient tcpClient = options != null ? vertx.createNetClient(options) : vertx.createNetClient();

// Connect to the memcached instance
Future<NetSocket> connect = tcpClient.connect(port, host);
return connect.map(so -> {
  // Create the client
  MemcachedClientImpl memcachedClient = new MemcachedClientImpl((VertxInternal) vertx, (NetSocketInternal) so);

  // Initialize the client: configure the pipeline and set the handlers
  memcachedClient.init();

  return memcachedClient;
});

ChannelPipeline pipeline = so.channelHandlerContext().pipeline();

// Add the memcached message aggregator
pipeline.addFirst("aggregator", new BinaryMemcacheObjectAggregator(Integer.MAX_VALUE));

// Add the memcached decoder
pipeline.addFirst("memcached", new BinaryMemcacheClientCodec());

// Set the message handler to process memcached message
so.messageHandler(this::processResponse);

return so.writeMessage(request).compose(v -> {

  // The message has been encoded successfully and sent
  // Create a response promise and add it to the inflight queue, so it can be resolved by the server ack
  Promise<FullBinaryMemcacheResponse> promise = vertx.promise();
  inflight.add(promise);

  //
  return promise.future();
});

FullBinaryMemcacheResponse response = (FullBinaryMemcacheResponse) msg;

try {
  // Get the handler that will process the response
  Promise<FullBinaryMemcacheResponse> handler = inflight.poll();

  // Handle the message
  handler.complete(response);
} finally {
  // Release the referenced counted message
  response.release();
}

ByteBuf keyBuf = Unpooled.copiedBuffer(key, StandardCharsets.UTF_8);

// Create the memcached request
FullBinaryMemcacheRequest request = new DefaultFullBinaryMemcacheRequest(keyBuf, Unpooled.EMPTY_BUFFER);

// Set the memcached operation opcode to perform a GET
request.setOpcode(BinaryMemcacheOpcodes.GET);

// Execute the request and process the response
return writeRequest(request).map(response -> processGetResponse(response));

short status = response.status();
switch (status) {

  case 0:
    // Succesfull get
    return response.content().toString(StandardCharsets.UTF_8);

  case 1:
    // Empty response -> null
    return null;

  default:
    // Memcached error
    throw new MemcachedError(status);
}

ByteBuf keyBuf = Unpooled.copiedBuffer(key, StandardCharsets.UTF_8);

// Create the memcached request
FullBinaryMemcacheRequest request = new DefaultFullBinaryMemcacheRequest(keyBuf, Unpooled.EMPTY_BUFFER);

// Set the memcached operation opcode to perform a GET
request.setOpcode(BinaryMemcacheOpcodes.GET);

// Execute the request and process the response
return writeRequest(request).map(response -> processGetResponse(response));

short status = response.status();
if (status == 0) {
  // Succesfull get
  return null;
} else {
  // Memcached error
  throw new MemcachedError(status);
}