Java：Thrift使用探秘

Love The Way You Lie 2022-09-29 11:59 59阅读 0赞

Thrift是apache提供的跨语言RPC调用框架，我们通过以下例子说明使用方法

tutorial.thrift：

enum Operation {
      ADD = 1,
      SUBTRACT = 2,
      MULTIPLY = 3,
      DIVIDE = 4
    }
    
    struct Work {
      1: i32 num1 = 0,
      2: i32 num2,
      3: Operation op,
      4: optional string comment,
    }
    
    service Calculator {
       i32 add(1:i32 num1, 2:i32 num2),
       i32 addList(1:list<Work> workList),
    }

通过如下命令：

set inDir=D:/project/build/example
    set outDir=D:/project/test
    
    D:/programs/thrift-0.10.0.exe -r --gen java -out %outDir% %inDir%/tutorial.thrift

将生成如下java文件：  
Work.java  
Operation.java  
Calculator.java

下面是测试代码：

public class Client{
        public static void main(String[] args) {
            try {
                TTransport transport = new TSocket("localhost", 9090);
    
                TProtocol protocol = new TBinaryProtocol(transport);
                Calculator.Client client = new Calculator.Client(protocol);
    
                transport.open();
    
                performs(client);
                
                transport.close();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        
        public static void performs(Calculator.Client client){
            Work work1 = new Work();
    
            work1.op = Operation.DIVIDE;
            work1.num1 = 1;
            work1.num2 = 0;
    
            Work work2 = new Work();
    
            work2.op = Operation.SUBTRACT;
            work2.num1 = 15;
            work2.num2 = 10;
    
            List<Work> workList = new ArrayList<Work>();
            workList.add(work1);
            workList.add(work2);
    
            int r1 = client.addList(workList);
            System.out.println("addList=" + r1);
            
            int retAdd = client.add(1, 2);
            System.out.println("retAdd=" + retAdd);
        }
    }

写一个CalculatorHandler实现Calculator.Iface

public class Server{
        public static void simple() {
            try {
                CalculatorHandler handler = new CalculatorHandler();
                Calculator.Processor<CalculatorHandler> processor = new Calculator.Processor<CalculatorHandler>(handler);
                
                TServerTransport serverTransport = new TServerSocket(9090);
                TServer server = new TSimpleServer(new Args(serverTransport).processor(processor));
    
                server.serve();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

将org.apache.thrift.protocol.TBinaryProtocol中所有read和write方法都加上注释，我们看看运行时的报文结构是怎样的  
write是返回报文结构和read一样，所以就不再注释了

--readI32:-2147418111  版本号
    --readI32:7            addList的字符长度
    --readString:addList
    --readI32:1            addList是本次调用的第一个方法
    --byte:15              方法addList的第一个元素的类型是TType.LIST
    --readI16:1            方法addList的第一个元素的序号
    --byte:12              TType.STRUCT，这里指Work
    --readI32:2            TType.LIST对象中Work对象的个数
    --byte:8               第一个Work对象中第一个元素的类型TType.I32
    --readI16:1            第一个Work对象中第一个元素的序号
    --readI32:1            第一个Work对象中第一个元素的值
    --byte:8               第一个Work对象中第二个元素的类型TType.I32
    --readI16:2            第一个Work对象中第二个元素的序号
    --readI32:0            第一个Work对象中第二个元素的值
    --byte:8               第一个Work对象中第三个元素的类型TType.I32
    --readI16:3            第一个Work对象中第三个元素的序号
    --readI32:4            第一个Work对象中第三个元素的值Operation.DIVIDE
    --byte:0               第一个Work对象结束TType.STOP
    --byte:8               第二个Work对象中第一个元素的类型TType.I32
    --readI16:1            第二个Work对象中第一个元素的序号
    --readI32:15           第二个Work对象中第一个元素的值
    --byte:8               第二个Work对象中第二个元素的类型TType.I32
    --readI16:2            第二个Work对象中第二个元素的序号
    --readI32:10           第二个Work对象中第二个元素的值
    --byte:8               第二个Work对象中第三个元素的类型TType.I32
    --readI16:3            第二个Work对象中第三个元素的序号
    --readI32:2            第二个Work对象中第三个元素的值Operation.SUBTRACT
    --byte:0               第二个Work对象结束TType.STOP
    --byte:0               方法addList参数结束TType.STOP
    
    --writeI32:-2147418110
    --writeI32:7
    --writeString:addList
    --writeI32:1
    --writeByte:8
    --writeI16:0
    --writeI32:16
    --writeByte:0
    
    --readI32:-2147418111  版本号
    --readI32:3            add的字符长度
    --readString:add
    --readI32:2            add是本次调用的第二个方法
    --readByte:8           第一个元素的类型是TType.I32
    --readI16:1            第一个元素的序号
    --readI32:1            第一个元素的值
    --readByte:8           第二个元素的类型TType.I32
    --readI16:2            第二个元素的序号
    --readI32:2            第二个元素的值
    --readByte:0           方法add参数结束TType.STOP
    
    --writeI32:-2147418110
    --writeI32:3
    --writeString:add
    --writeI32:2
    --writeByte:8
    --writeI16:0
    --writeI32:3
    --writeByte:0

上面的例子只能用于一个接口Calculator，如果还有一个接口Sender该怎么办？

service Sender {
           void ping(),
    }

可以使用TMultiplexedProcessor作处理器:

///server//
    TMultiplexedProcessor processor = new TMultiplexedProcessor();
    
    processor.registerProcessor("Calculator", new Calculator.Processor<CalculatorHandler>(new CalculatorHandler()));
    processor.registerProcessor("Sender", new Ping.Processor<SenderHandler>(new SenderHandler()));
    
    TServerTransport t = new TServerSocket(9090);
    
    Args args = new Args(t);
    args.processor(processor);
    
    TSimpleServer server =new TSimpleServer(args);

///client//
    TTransport transport =  new TSocket("localhost", 9090) ;
    transport.open();
    
    TProtocol protocol1 = new TMultiplexedProtocol(new TBinaryProtocol(transport), "Calculator");
    Calculator.Client client1 = new Calculator.Client(protocol1);
    
    TProtocol protocol2 = new TMultiplexedProtocol(new TBinaryProtocol(transport), "Sender");
    Sender.Client client2 = new Sender.Client(protocol2);

上面例子中TProtocol用的是TBinaryProtocol，这种报文虽然是二进制的，但报文还是太大，可以使用TCompactProtocol替换TBinaryProtocol，这样可以将报文压缩  
  
上面例子中TTransport用的是TSocket，这种报文结构没有说明报文长度，无法与netty/mina等第三方IO框架整合，可以使用TFramedTransport，它会在原报文前加上4个字节，并在其中写入其后的报文长度，这样就方便使用了：

TTransport transport = new TFramedTransport(new TSocket("localhost", 9090));

如果要服务器端使用NIO，异常多线程处理请求，可以使用如下代码：

try {
    	CalculatorHandler handler = new CalculatorHandler();
    	Calculator.Processor<CalculatorHandler> processor = new Calculator.Processor<CalculatorHandler>(handler);
    
    	TNonblockingServerSocket serverTransport = new TNonblockingServerSocket(9090);
    
    	THsHaServer.Args serverArgs = new THsHaServer.Args(serverTransport);
    	serverArgs.transportFactory(new TFramedTransport.Factory());
    	serverArgs.protocolFactory(new TBinaryProtocol.Factory());
    	serverArgs.maxWorkerThreads(10);
    	serverArgs.processor(processor);
    
    	TServer server = new THsHaServer(serverArgs);
    
    	server.serve();
    } catch (Exception e) {
    	e.printStackTrace();
    }