无阻塞的thrift远程调用demo

概述

在研究如何利用thrift开发高扩展、高并发、高容错的rpc框架的学习过程中有了本篇文章做备忘。例子中使用的通信方式在进行进一步封装扩展后就完全可以用于生产环境。本例并非常见的简单例子。

接口IDL

namespace java cn.test.rpc.api
namespace cpp cn.test.rpc.api

struct TResponse {
    1: required i32 uid;
    2: optional string userName;
}
struct TRequest {
    1: required i32 uid;
    2: optional string userName;
}
service HelloWorld{
    TResponse sayHello(1:TRequest tRequest);
}

整个rpc过程目前只在java范围内，所以IDL的编译交给了maven插件来做，方便快捷。如下配置：

• 依赖

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

• 插件

  <plugin>
    <groupId>org.apache.thrift.tools</groupId>
    <artifactId>maven-thrift-plugin</artifactId>
    <version>0.1.11</version>
    <executions>
        <execution>
            <id>thrift-sources</id>
            <phase>generate-sources</phase>
            <goals>
                <goal>compile</goal>
            </goals>
        </execution>
        <execution>
            <id>thrift-test-sources</id>
            <phase>generate-test-sources</phase>
            <goals>
                <goal>testCompile</goal>
            </goals>
        </execution>
    </executions>
  </plugin>

Server端实现

public class BootStrap {
    public static void main(String[] args) throws TTransportException, UnknownHostException {
        HelloWorld.Processor processor = new HelloWorld.Processor<IHelloWord>(new HelloWorldImpl());
        InetAddress addr = InetAddress.getLocalHost();
        String ip= addr.getHostAddress();//获得本机IP
        TNonblockingServerTransport st = new TNonblockingServerSocket(new InetSocketAddress(ip,9999));
        TNonblockingServer.Args trArgs=new TNonblockingServer.Args(st);
        trArgs.processor(processor);
        trArgs.protocolFactory(new TBinaryProtocol.Factory(true, true));
        TNonblockingServer server = new TNonblockingServer(trArgs);
        server.serve();
    }
}

Client端实现

public class ClientTNio {
    public static void main(String[] args) throws TException {
        Runnable task = new Runnable() {
            @Override
            public void run() {
                try {
                    TSocket socket = new TSocket("10.10.8.8", 9999); //启动服务的那台机器
                    TTransport transport = new TFramedTransport(socket);
                    transport.open();
                    TProtocol protocol = new TBinaryProtocol(transport);
                    HelloWorld.Client client = new HelloWorld.Client(protocol);
                    TRequest r = new TRequest();
                    r.setUid(1500);
                    r.setUserName("tt");
                    TResponse p = client.sayHello(r);
                    transport.close();
                    System.out.println(p);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        };
        Thread[] pool = new Thread[1];
        for (int i =0;i<1;i++){
            pool[i] = new Thread(task);
            pool[i].start();
        }
    }
}

结果测试

• Server端结果

  TRequest(uid:1500, userName:tt)

• client端结果

  TResponse(uid:1500, userName:tt remote)