Arrow Flight RPC #

Arrow Flight 是一个 RPC 框架，用于在网络上高效传输 Arrow 数据。

另请参阅

Flight 协议文档: Flight 协议的文档，包括如何在概念上使用 Flight。
Java 食谱: 在 Java 中使用 Arrow Flight 的食谱。

编写 Flight 服务 #

Flight 服务器实现 FlightProducer 接口。为了方便起见，它们可以继承 NoOpFlightProducer，它提供了所有 RPC 方法的默认实现。

public class TutorialFlightProducer implements FlightProducer {
    @Override
    // Override methods or use NoOpFlightProducer for only methods needed
}

每个 RPC 方法总是接受一个 CallContext 来获取通用参数。若要指示失败，请将异常传递给（如果存在）“侦听器”，否则引发异常。

// Server
@Override
public void listFlights(CallContext context, Criteria criteria, StreamListener<FlightInfo> listener) {
    // ...
    listener.onError(
        CallStatus.UNAUTHENTICATED.withDescription(
            "Custom UNAUTHENTICATED description message.").toRuntimeException());
    // ...
}

// Client
try{
    Iterable<FlightInfo> flightInfosBefore = flightClient.listFlights(Criteria.ALL);
    // ...
} catch (FlightRuntimeException e){
    // Catch UNAUTHENTICATED exception
}

若要启动服务器，请创建一个 Location 来指定监听位置，然后使用生产者的实例创建一个 FlightServer。这将启动服务器，但不会阻塞程序的其余部分。调用 FlightServer.awaitTermination 来阻塞直到服务器停止。

class TutorialFlightProducer implements FlightProducer {
    @Override
    // Override methods or use NoOpFlightProducer for only methods needed
}

Location location = Location.forGrpcInsecure("0.0.0.0", 0);
try(
    BufferAllocator allocator = new RootAllocator();
    FlightServer server = FlightServer.builder(
            allocator,
            location,
            new TutorialFlightProducer()
    ).build();
){
    server.start();
    System.out.println("Server listening on port " + server.getPort());
    server.awaitTermination();
} catch (Exception e) {
    e.printStackTrace();
}

Server listening on port 58104

使用 Flight 客户端 #

要连接到 Flight 服务，请使用位置创建一个 FlightClient。

Location location = Location.forGrpcInsecure("0.0.0.0", 58104);

try(BufferAllocator allocator = new RootAllocator();
    FlightClient client = FlightClient.builder(allocator, location).build()){
    // ... Consume operations exposed by Flight server
} catch (Exception e) {
    e.printStackTrace();
}

取消和超时 #

进行调用时，客户端可以选择提供 CallOptions。这允许客户端为调用设置超时时间。此外，客户端 RPC 调用返回的一些对象公开了取消方法，允许提前终止调用。

Location location = Location.forGrpcInsecure("0.0.0.0", 58609);

try(BufferAllocator allocator = new RootAllocator();
    FlightClient tutorialFlightClient = FlightClient.builder(allocator, location).build()){

    Iterator<Result> resultIterator = tutorialFlightClient.doAction(
            new Action("test-timeout"),
            CallOptions.timeout(2, TimeUnit.SECONDS)
    );
} catch (Exception e) {
    e.printStackTrace();
}

在服务器端，超时时间是透明的。对于取消，服务器需要手动轮询 setOnCancelHandler 或 isCancelled 以检查客户端是否已取消调用，如果是，则退出服务器当前正在执行的任何处理。

// Client
Location location = Location.forGrpcInsecure("0.0.0.0", 58609);
try(BufferAllocator allocator = new RootAllocator();
    FlightClient tutorialFlightClient = FlightClient.builder(allocator, location).build()){
    try(FlightStream flightStream = flightClient.getStream(new Ticket(new byte[]{}))) {
        // ...
        flightStream.cancel("tutorial-cancel", new Exception("Testing cancellation option!"));
    }
} catch (Exception e) {
    e.printStackTrace();
}
// Server
@Override
public void getStream(CallContext context, Ticket ticket, ServerStreamListener listener) {
    // ...
    listener.setOnCancelHandler(()->{
                // Implement logic to handle cancellation option
            });
}

启用 TLS #

通过向 FlightServer.Builder.useTls 提供证书和密钥对，可以在设置服务器时启用 TLS。

在客户端，使用 Location.forGrpcTls 为客户端创建位置。

启用身份验证 #

警告

在没有启用 TLS 的情况下，身份验证是不安全的。

通过实现 ServerAuthHandler 可以启用基于握手身份验证。身份验证包括两个部分：在初始客户端连接时，服务器和客户端身份验证实现可以执行任何必要的协商。然后，客户端身份验证处理程序将提供一个令牌，该令牌将附加到未来的调用中。

客户端通过 ClientAuthHandler.authenticate 发送要验证的数据。服务器通过 ServerAuthHandler.authenticate 验证收到的数据。

自定义中间件 #

服务器和客户端支持自定义中间件（或拦截器），它们在每个请求时调用，并可以有限地修改请求。这些可以通过实现 FlightServerMiddleware 和 FlightClientMiddleware 接口来实现。

中间件相当有限，但它们可以向请求/响应添加标题。在服务器上，它们可以检查传入的标题并使请求失败；因此，它们可以用来实现自定义身份验证方法。

添加服务 #

服务器可以添加其他 gRPC 服务。例如，要添加健康检查服务

final HealthStatusManager statusManager = new HealthStatusManager();
final Consumer<NettyServerBuilder> consumer = (builder) -> {
  builder.addService(statusManager.getHealthService());
};
final Location location = forGrpcInsecure(LOCALHOST, 5555);
try (
    BufferAllocator a = new RootAllocator(Long.MAX_VALUE);
    Producer producer = new Producer(a);
    FlightServer s = FlightServer.builder(a, location, producer)
        .transportHint("grpc.builderConsumer", consumer).build().start();
) {
  Channel channel = NettyChannelBuilder.forAddress(location.toSocketAddress()).usePlaintext().build();
  HealthCheckResponse response = HealthGrpc
          .newBlockingStub(channel)
          .check(HealthCheckRequest.getDefaultInstance());

  System.out.println(response.getStatus());
}