使用 TFramedTransport 时出现 TTransportException

Question

我对这个问题很困惑。我有一个 Apache Thrift 0.9.0 客户端和服务器。客户端代码如下：

this.transport = new TSocket(this.server, this.port);
final TProtocol protocol = new TBinaryProtocol(this.transport);
this.client = new ZKProtoService.Client(protocol);

这很好用。但是，如果我尝试将传输包装在 TFramedTransport

this.transport = new TSocket(this.server, this.port);
final TProtocol protocol = new TBinaryProtocol(new TFramedTransport(this.transport));
this.client = new ZKProtoService.Client(protocol);

我在客户端收到以下模糊（没有任何解释消息）异常。服务器端显示没有错误。

org.apache.thrift.transport.TTransportException
    at org.apache.thrift.transport.TIOStreamTransport.read(TIOStreamTransport.java:132)
    at org.apache.thrift.transport.TTransport.readAll(TTransport.java:84)
    at org.apache.thrift.transport.TFramedTransport.readFrame(TFramedTransport.java:129)
    at org.apache.thrift.transport.TFramedTransport.read(TFramedTransport.java:101)
    at org.apache.thrift.transport.TTransport.readAll(TTransport.java:84)
    at org.apache.thrift.protocol.TBinaryProtocol.readAll(TBinaryProtocol.java:378)
    at org.apache.thrift.protocol.TBinaryProtocol.readI32(TBinaryProtocol.java:297)
    at org.apache.thrift.protocol.TBinaryProtocol.readMessageBegin(TBinaryProtocol.java:204)
    at org.apache.thrift.TServiceClient.receiveBase(TServiceClient.java:69)
    at com.blablabla.android.core.device.proto.ProtoService$Client.recv_open(ProtoService.java:108)
    at com.blablabla.android.core.device.proto.ProtoService$Client.open(ProtoService.java:95)
    at com.blablabla.simpleprotoclient.proto.ProtoClient.initializeCommunication(ProtoClient.java:411)
    at com.blablabla.simpleprotoclient.proto.ProtoClient.doWork(ProtoClient.java:269)
    at com.blablabla.simpleprotoclient.proto.ProtoClient.run(ProtoClient.java:499)
    at java.lang.Thread.run(Thread.java:724)

如果我使用TCompactProtocol 而不是TBinaryProtocol，它也会失败。

在服务器端，我使用自己的类扩展了TProcessor，因为我需要为此客户端重用现有的服务处理程序（服务服务器端IFace 实现）：

@Override
public boolean process(final TProtocol in, final TProtocol out)
        throws TException {
    final TTransport t = in.getTransport();
    final TSocket socket = (TSocket) t;
    socket.setTimeout(ProtoServer.SOCKET_TIMEOUT);
    final String clientAddress = socket.getSocket().getInetAddress()
            .getHostAddress();
    final int clientPort = socket.getSocket().getPort();
    final String clientRemote = clientAddress + ":" + clientPort;
    ProtoService.Processor<ProtoServiceHandler> processor = PROCESSORS
            .get(clientRemote);
    if (processor == null) {
        final ProtoServiceHandler handler = new ProtoServiceHandler(
                clientRemote);
        processor = new ProtoService.Processor<ProtoServiceHandler>(
                handler);
        PROCESSORS.put(clientRemote, processor);
        HANDLERS.put(clientRemote, handler);
        ProtoClientConnectionChecker.addNewConnection(clientRemote,
                socket);
    }
    return processor.process(in, out);
}

这就是我启动服务器端的方式：

TServerTransport serverTransport = new TServerSocket(DEFAULT_CONTROL_PORT);
TServer server = new TThreadPoolServer(new TThreadPoolServer.Args(
            serverTransport).processor(new ControlProcessor()));
Thread thControlServer = new Thread(new StartServer("Control", server));
thControlServer.start();

我有一些问题：

• 重用服务处理程序实例是否正确，或者我不应该这样做？
• 为什么我使用TFramedTransport 或TCompactProtocol 会失败？如何解决这个问题？

欢迎对此问题提供任何帮助。提前致谢！

Answer 1

我遇到了同样的问题，终于找到了答案。可以在服务器上设置传输类型，尽管我在网上找到的大多数教程和示例都不清楚这一点。查看TServer.Args 类的所有方法（或其他服务器的args 类，它们扩展了TServer.Args）。有inputTransportFactory 和outputTransportFactory 方法。您可以使用new TFramedTransport.Factory() 作为这些方法的输入来声明服务器应该使用哪个传输。在斯卡拉中：

  val handler = new ServiceStatusHandler
  val processor = new ServiceStatus.Processor(handler)
  val serverTransport = new TServerSocket(9090)
  val args = new TServer.Args(serverTransport)
    .processor(processor)
    .inputTransportFactory(new TFramedTransport.Factory)
    .outputTransportFactory(new TFramedTransport.Factory)
  val server = new TSimpleServer(args)
  println("Starting the simple server...")
  server.serve()

请注意，如果您使用的是TAsyncClient，则您无法选择使用的传输方式。您必须使用TNonblockingTransport，它只有一个标准实现TNonblockingSocket，它在内部包装了您在帧传输中使用的任何协议。它实际上并没有将您选择的协议包装在TFramedTransport 中，但它确实将帧的长度添加到它写入的内容中，并希望服务器也将响应的长度添加到前面。这在我找到的任何地方都没有记录，但是如果您查看源代码并尝试不同的组合，您会发现TSimpleServer 必须使用TFramedTransport 才能使其与异步客户端一起使用。

顺便说一句，还值得注意的是，文档说TNonblockingServer 必须在传输的最外层使用TFramedTransport。但是，示例并未显示在TNonblockingServer.Args 中设置了此内容，但您仍然发现必须在客户端使用TFramedTransport 才能在服务器上成功执行rpc。这是因为TNonblockingServer.Args 的输入和输出协议默认设置为TFramedTransport（您可以使用反射来检查超类层次结构的字段或在AbstractNonblockingServerArgs 的构造函数的源代码中看到这一点——您可以覆盖输入和输出传输，但由于文档中讨论的原因，服务器可能会失败）。

Answer 2

如果问题发生在有框架的情况下，但它在没有框架的情况下也能正常工作，那么说明两端的协议栈不兼容。选择以下选项之一：

• 或者修改服务器代码以使用框架
• 或者不要在客户端使用框架

一个好的经验法则是，始终在两端使用完全相同的协议/传输堆栈。在特定情况下它会爆炸，因为 framed 添加了一个四字节的标头来保存大小的消息。如果服务器不使用框架，则客户端发送的这四个额外字节将被（错误地）解释为消息的一部分。

尽管该答案中的示例代码 TNonblockingServer in thrift crashes when TFramedTransport opens是针对C++的，在服务器上加框架应该和Java很相似。

PS：是的，重复使用你的处理程序是完全可以的。典型的处理程序是无状态的。