thrift 使用

超、凢脫俗 2022-05-27 01:38 230阅读 0赞

thrift也是一个rpc调用框架,在这里写点自己的经验,帮助自己复习.

1.支持的协议:
          TBinaryProtocol - 一种直接的二进制格式，将数值编码为二进制，而不是转换为文本。
          TCompactProtocol - 非常高效，密集的数据编码（详见下文）。
          TDenseProtocol - 类似于TCompactProtocol，但从传输内容中除去元信息，并将其添加回接收器。TDenseProtocol仍然是实验性的，并且在Java实现中尚不可用。
          TJSONProtocol - 使用JSON编码数据。
          TSimpleJSONProtocol - 使用JSON的只写协议。适用于通过脚本语言进行分析
          TDebugProtocol - 使用人类可读的文本格式来帮助调试。
    2.传输方式:
          TSocket - 使用阻塞套接字I / O进行传输。
          TFramedTransport - 以帧为单位发送数据，其中每个帧的前面都有一个长度。使用非阻塞服务器时需要此传输。
          TFileTransport - 此传输写入文件。虽然这个传输不包含在Java实现中，但它应该足够简单来实现。
          TMemoryTransport - 为I / O使用内存。Java实现在内部使用一个简单的ByteArrayOutputStream。
          TZlibTransport - 使用zlib执行压缩 。与另一个运输工具一起使用。在Java实现中不可用。
    3.thrift提供的服务器
          TSimpleServer - 使用std blocking io的单线程服务器。用于测试。
          TThreadPoolServer - 一个使用std blocking io的多线程服务器。
          TNonblockingServer - 使用非阻塞io的多线程服务器（Java实现使用NIO通道）。TFramedTransport必须与此服务器一起使用.
    * 半同步半异步的服务端模型，需要指定为： TFramedTransport 数据传输的方式  THsHaServer服务模型
    * 使用非阻塞式IO，服务端和客户端需要指定 TFramedTransport 数据传输的方式    TNonblockingServer 服务模型

一 服务端

1.简单的线程服务模型,一般用于测试

public class SimpleServerDemo {
        static int SERVER_PORT = 8888;
    
        public static void main(String[] args) throws TTransportException {
            TProcessor tprocessor = new HelloWorldService.Processor <HelloWorldService.Iface>(
                    new HelloHandler());
    
            TServerSocket serverTransport = new TServerSocket(SERVER_PORT);
    
            TServer.Args tArgs = new TServer.Args(serverTransport);
    
            tArgs.processor(tprocessor);
            tArgs.protocolFactory(new TBinaryProtocol.Factory());
            TServer server = new TSimpleServer(tArgs);
            server.serve();
    
        }
    
    } 2

2.线程池模型阻塞式

public class ThreadPoolServerDemo {
    
        private static final int SERVER_PORT = 8888;
    
        public static void main(String[] args) throws TTransportException {
            TProcessor ifaceTProcessor = new HelloWorldService.Processor <HelloWorldService.Iface>(
                    new HelloHandler()
            );
    
            //只返回一个单例
            TProcessorFactory tProcessorFactory = new TProcessorFactory(ifaceTProcessor);
    
            TServerSocket tServerSocket = new TServerSocket(SERVER_PORT);
            //线程池服务模型
            TThreadPoolServer.Args args1 = new TThreadPoolServer.Args(tServerSocket);
            args1.processorFactory(tProcessorFactory);
    
            //传输协议
            args1.protocolFactory(new TBinaryProtocol.Factory());
    
            //启动服务
            TServer server = new TThreadPoolServer(args1);
            server.serve();
        }
    }

3.半同步半异步服务端模型,需要指定TFramedTransport 数据传输方式,THsHaServer服务模型

public class THsHaServerDemo {
    
        private static final int SERVER_PORT = 8888;
    
        public static void main(String[] args) throws TTransportException {
            //创建处理器
            TProcessor tProcessor = new HelloWorldService.Processor<HelloWorldService.Iface>(
                    new HelloHandler()
            );
    
            //创建连接模型
            TNonblockingServerSocket tNonblockingServerSocket = new TNonblockingServerSocket(SERVER_PORT);
    
            //参数
            THsHaServer.Args args1 = new THsHaServer.Args(tNonblockingServerSocket);
            args1.processor(tProcessor);
            args1.transportFactory(new TFramedTransport.Factory()); //传输方式
            args1.protocolFactory(new TBinaryProtocol.Factory()); //协议
    
            TServer server = new THsHaServer(args1);
            server.serve();
        }
    }

4.非阻塞式IO FramedTransport 数据传输方式,TNonblockingServer服务模型

public class TNonblockingServerDemo {
    
        private final static int SERVER_PORT = 8888;
    
    
        public static void main(String[] args) throws TTransportException {
    
            //创建执行器
    
            TProcessor tProcessor = new HelloWorldService.Processor<HelloWorldService.Iface>(
                    new HelloHandler()
            );
    
            //创建服务模型
            TNonblockingServerSocket tNonblockingServerSocket = new TNonblockingServerSocket(SERVER_PORT);
    
            //设置参数
            TNonblockingServer.Args args1 = new TNonblockingServer.Args(tNonblockingServerSocket);
            args1.processor(tProcessor);
            args1.transportFactory(new TFramedTransport.Factory()); //执行传输方式
            args1.protocolFactory(new TCompactProtocol.Factory()); //TODO 传输协议 //启动
    
            TNonblockingServer tNonblockingServer = new TNonblockingServer(args1);
    
            tNonblockingServer.serve();
        }
    }

**PS:上面所说的要指定的服务模型与传输方式都是需要同时指定服务端与客户端的!**

二 客户端

1.同步客户端

public class ClientDemo {
        public static final String SERVER_IP = "localhost";
        private static final int SERVER_PORT = 8888;
        public static final int TIMEOUT = 30000;
    
    
        public static void main(String[] args) {
    
            simpleClientDemo();
        }
    
    
        /** * 此方法为需要注意的是协议与服务端一致 */ private static void simpleClientDemo() {
            TTransport tTransport = null;
    
            try {
                //TODO 传输方式也要一致 //tTransport = new TSocket(SERVER_IP, SERVER_PORT, TIMEOUT); //服务端没有指定ip so 默认为本地
                tTransport = new TFramedTransport(new TSocket(SERVER_IP, SERVER_PORT, TIMEOUT));
                //TODO 协议要与服务端一致 //TProtocol tBinaryProtocol = new TBinaryProtocol(tTransport);
                TProtocol tBinaryProtocol = new TBinaryProtocol(tTransport);
    
                //
                HelloWorldService.Client client = new HelloWorldService.Client(tBinaryProtocol);
    
                //
                tTransport.open();
                //调用方法
                String callback = client.sayHello("username");
    
                System.out.println(callback);
    
            } catch (TTransportException e) {
                e.printStackTrace();
            } catch (TException e) {
                e.printStackTrace();
            } finally {
                if (null != tTransport) {
                    tTransport.close();
                }
            }
        }
    }

2.异步客户端

public class AsynClientDemo {
    
        public static final String SERVER_IP = "localhost";
        public static final int SERVER_PORT = 8888;
        public static final int TIMEOUT = 30000;
    
    
        public static void main(String[] args) {
            try {
                //异步管理器
                TAsyncClientManager clientManager = new TAsyncClientManager();
                //tNonblockingTransport 服务模型
                TNonblockingTransport tNonblockingTransport = new TNonblockingSocket(SERVER_IP, SERVER_PORT, TIMEOUT);
                //协议工厂
                TProtocolFactory tProcessorFactory = new TCompactProtocol.Factory();
                //TODO 异步客户端 参数: 协议,管理器,协议工厂 HelloWorldService.AsyncClient asyncClient = new HelloWorldService.AsyncClient(tProcessorFactory, clientManager, tNonblockingTransport);
    
                System.out.println("------------------------开始--------------------------");
    
                //
                CountDownLatch latch = new CountDownLatch(1);
                AsynCallback callBack = new AsynCallback(latch);
    
                asyncClient.sayHello("username", callBack);
                System.out.println("call method sayHello .... end");
                boolean wait = latch.await(30, TimeUnit.SECONDS);
                System.out.println("latch.await =:" + wait);
    
            } catch (IOException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (TException e) {
                e.printStackTrace();
            }
        }
    
        /** * 回调函数 */ public static class AsynCallback implements AsyncMethodCallback <String> {
            private CountDownLatch latch;
    
            public AsynCallback(CountDownLatch latch) {
                this.latch = latch;
            }
    
            /** * This method will be called when the remote side has completed invoking * your method call and the result is fully read. For { @code oneway} method * calls, this method will be called as soon as we have completed writing out * the request. * * @param response The return value of the asynchronously invoked method; * { @code null} for void methods which includes * { @code oneway} methods. */ @Override public void onComplete(String response) {
                System.out.println("onComplete");
                try {
                    Thread.sleep(1000L * 1);
                    System.out.println("AsynCall result =:" + response);
                } catch (Exception e) {
                    e.printStackTrace();
                } /*finally {
                    latch.countDown();
                }*/
            }
    
            public void onError(Exception exception) {
                System.out.println("onError :" + exception.getMessage());
                //latch.countDown();
            }
        }
    }

三 thrift文件内容

namespace java com.example.thrift.gen
    //这是服务(也就是方法)
    service HelloWorldService {
       string sayHello(1:string username)
    }

四 pom文件

<dependency>
        <groupId>org.apache.thrift</groupId>
        <artifactId>libthrift</artifactId>
        <version>0.10.0</version>
    </dependency>

<build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
    
            <plugin>
                <groupId>org.apache.thrift</groupId>
                <artifactId>thrift-maven-plugin</artifactId>
                <version>0.10.0</version>
                <configuration>
                    <thriftExecutable>E:/MySoftWare/MyTools/thrift/thrift-0.10.0.exe</thriftExecutable>
                    <thriftSourceRoot>${basedir}/src/main/resources/thrift/idl/</thriftSourceRoot>
                    <outputDirectory>${basedir}/src/main/java</outputDirectory>
                </configuration>
                <executions>
                    <execution>
                        <id>thrift-sources</id>
                        <phase>generate-sources</phase>
                        <goals>
                            
                        </goals>
                    </execution>
                    
                </executions>
            </plugin>
    
        </plugins>
    </build>

**PS:thriftExecutable 值为thrift编译工具的位置**

五 handler 文件

public class HelloHandler implements HelloWorldService.Iface {
        @Override public String sayHello(String username) throws TException {
            System.out.println("收到的信息:" + username);
            return "hello ----------------->" + username;
        }
    }

**PS:如有问题,请给我留言!**

参考:[https://blog.csdn.net/lk10207160511/article/details/50450541][https_blog.csdn.net_lk10207160511_article_details_50450541]

[https_blog.csdn.net_lk10207160511_article_details_50450541]: https://blog.csdn.net/lk10207160511/article/details/50450541