Thrift 基本使用

数据类型

1. 基本类型：
   
   • bool：布尔值，true 或 false，对应 Java 的 boolean
   • byte：8 位有符号整数，对应 Java 的 byte
   • i16：16 位有符号整数，对应 Java 的 short
   • i32：32 位有符号整数，对应 Java 的 int
   • i64：64 位有符号整数，对应 Java 的 long
   • double：64 位浮点数，对应 Java 的 double
   • string：utf-8编码的字符串，对应 Java 的 String
2. 结构体类型：
   
   • struct：定义公共的对象，类似于 C 语言中的结构体定义，在 Java 中是一个 JavaBean
3. 容器类型：
   
   • list：对应 Java 的 ArrayList
   • set：对应 Java 的 HashSet
   • map：对应 Java 的 HashMap
4. 异常类型：
   
   • exception：对应 Java 的 Exception
5. 服务类型：
   
   • service：对应服务的类

编写步骤

1. 服务端：

   • 实现服务处理接口impl
   • 创建TProcessor
   • 创建TServerTransport
   • 创建TProtocol
   • 创建TServer
   • 启动Server

2. 客户端

   • 创建Transport
   • 创建TProtocol
   • 基于TTransport和TProtocol创建 Client
   • 调用Client的相应方法

3. 传输协议

   • TBinaryProtocol : 二进制格式.
   • TCompactProtocol : 压缩格式
   • TJSONProtocol : JSON格式
   • TSimpleJSONProtocol : 提供JSON只写协议, 生成的文件很容易通过脚本语言解析

传输方式

• TSocket- 使用堵塞式I/O进行传输，也是最常见的模式。
• TFramedTransport- 使用非阻塞方式，按块的大小，进行传输，类似于Java中的NIO。
• TFileTransport- 顾名思义按照文件的方式进程传输，虽然这种方式不提供Java的实现，但是实现起来非常简单。
• TMemoryTransport- 使用内存I/O，就好比Java中的ByteArrayOutputStream实现。
• TZlibTransport- 使用执行zlib压缩，不提供Java的实现。

服务端类型

• TSimpleServer - 单线程服务器端使用标准的堵塞式I/O。
• TThreadPoolServer - 多线程服务器端使用标准的堵塞式I/O。
• TNonblockingServer – 多线程服务器端使用非堵塞式I/O，并且实现了Java中的NIO通道。

实例演示 - 准备工作

1. 创建脚本

创建文件demoHello.thrift

namespace java com.fengtang.thrift.demo //定义生成代码的命名空间，与package相对应。
//代码生成的类名，你的业务逻辑代码需要实现代码生成的HelloWorldService.Iface接口
service  HelloWorldService {
    //方法名称和方法中的入参
    string sayHello(1:string username)
}

1. 生成代码

使用thrift-0.9.2.exe 生成代码，执行命令

thrift-0.9.2.exe -r -gen java ./demoHello.thrift

生成后的截图

3.将代码拷入IDEA

使用maven解决依赖关系

        <dependency>
            <groupId>org.apache.thrift</groupId>
            <artifactId>libthrift</artifactId>
            <version>0.9.2</version>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
            <version>1.5.8</version>
        </dependency>

实现接口Iface

package com.fengtang.thrift.demo.service;

import com.fengtang.thrift.demo.HelloWorldService;
import org.apache.thrift.TException;

/**
 * Create by fengtang
 * 2015/8/14 0014
 * Thrift
 */

public class HelloWorldImpl implements HelloWorldService.Iface {

    public HelloWorldImpl() {
    }
    @Override
    public String sayHello(String username) throws TException {
        return "Hi," + username + " I‘m the server ...";
    }

}

实例演示 - 分别测试不同类型的客户端和服务端模型

编写TSimpleServer服务端和客户端测试

TSimpleServer 简单的单线程服务模型，一般用于测试

服务端

package com.fengtang.thrift.demo.server;

import com.fengtang.thrift.demo.HelloWorldService;
import com.fengtang.thrift.demo.service.HelloWorldImpl;
import org.apache.thrift.TProcessor;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.server.TServer;
import org.apache.thrift.server.TSimpleServer;
import org.apache.thrift.transport.TServerSocket;

/**
 * Create by fengtang
 * 2015/8/14 0014
 * Thrift
 * TSimpleServer 简单的单线程服务模型
 */
public class TSimpServer {
    public static final int SERVER_PORT = 8090;

    /**
     * 创建TProcessor
     * 创建TServerTransport
     * 创建TProtocol
     * 创建TServer
     * 启动Server
     */
    public void startServer() {
        System.out.println("TSimpleServer start ....");

        try {
            //创建TProcessor
            TProcessor tprocessor = new HelloWorldService.Processor<HelloWorldService.Iface>(
                    new HelloWorldImpl());
            //创建TServerTransport
            TServerSocket serverTransport = new TServerSocket(SERVER_PORT);
            TServer.Args tArgs = new TServer.Args(serverTransport);
            tArgs.processor(tprocessor);
            tArgs.protocolFactory(new TBinaryProtocol.Factory());
            //创建TServer
            TServer server = new TSimpleServer(tArgs);
            server.serve();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    public static void main(String[] args) {
        TSimpServer server = new TSimpServer();
        server.startServer();
    }
}

客户端

package com.fengtang.thrift.demo.client;

import com.fengtang.thrift.demo.HelloWorldService;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;

/**
 * Create by fengtang
 * 2015/8/14 0014
 * Thrift
 */
public class TSimpleClient {
    public static final String SERVER_IP = "localhost";
    public static final int SERVER_PORT = 8090;
    public static final int TIMEOUT = 30000;

    /**
     * @param userName
     */
    public void startClient(String userName) {
        TTransport transport = null;
        try {
            transport = new TSocket(SERVER_IP, SERVER_PORT, TIMEOUT);
            // 协议要和服务端一致
            TProtocol protocol = new TBinaryProtocol(transport);
            HelloWorldService.Client client = new HelloWorldService.Client(protocol);
            transport.open();
            long start = System.currentTimeMillis();
            String result = client.sayHello(userName);
            System.out.println("Total time is " + (System.currentTimeMillis() - start) + " ms ");
            System.out.println("Thrift client result : \n" + result);
        } catch (TTransportException e) {
            e.printStackTrace();
        } catch (TException e) {
            e.printStackTrace();
        } finally {
            if (null != transport) {
                transport.close();
            }
        }
    }

    public static void main(String[] args) {
        TSimpleClient client = new TSimpleClient();
        client.startClient("Nicholas");
    }
}

运行结果截图

编写TThreadPoolServer服务端和客户端测试

TThreadPoolServer 线程池服务模型，使用标准的阻塞式IO，预先创建一组线程处理请求

服务端

package com.fengtang.thrift.demo.server;

import com.fengtang.thrift.demo.HelloWorldService;
import com.fengtang.thrift.demo.service.HelloWorldImpl;
import org.apache.thrift.TProcessor;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.server.TServer;
import org.apache.thrift.server.TThreadPoolServer;
import org.apache.thrift.transport.TServerSocket;

/**
 * Create by fengtang
 * 2015/8/14 0014
 * Thrift
 * 线程池服务模型，使用标准的阻塞式IO，预先创建一组线程处理请求。
 */
public class TThrPoolServer {
    public static final int SERVER_PORT = 8090;
    public void startServer() {
        try {
            System.out.println("HelloWorld TThreadPoolServer start ....");

            TProcessor tprocessor = new HelloWorldService.Processor<HelloWorldService.Iface>(
                    new HelloWorldImpl());

            TServerSocket serverTransport = new TServerSocket(SERVER_PORT);
            TThreadPoolServer.Args ttpsArgs = new TThreadPoolServer.Args(
                    serverTransport);
            ttpsArgs.processor(tprocessor);
            ttpsArgs.protocolFactory(new TBinaryProtocol.Factory());

            // 线程池服务模型，使用标准的阻塞式IO，预先创建一组线程处理请求。
            TServer server = new TThreadPoolServer(ttpsArgs);
            server.serve();
        } catch (Exception e) {
            System.out.println("Server start error!!!");
            e.printStackTrace();
        }
    }

    /**
     * @param args
     */
    public static void main(String[] args) {
        TThrPoolServer server = new TThrPoolServer();
        server.startServer();
    }
}

客户端

package com.fengtang.thrift.demo.client;

import com.fengtang.thrift.demo.HelloWorldService;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;

/**
 * Create by fengtang
 * 2015/8/14 0014
 * Thrift
 */
public class TThreadPoolClient {
    public static final String SERVER_IP = "localhost";
    public static final int SERVER_PORT = 8090;
    public static final int TIMEOUT = 30000;

    /**
     * @param userName
     */
    public void startClient(String userName) {
        TTransport transport = null;
        try {
            transport = new TSocket(SERVER_IP, SERVER_PORT, TIMEOUT);
            // 协议要和服务端一致
            TProtocol protocol = new TBinaryProtocol(transport);
            HelloWorldService.Client client = new HelloWorldService.Client(protocol);
            transport.open();
            long start = System.currentTimeMillis();
            String result = client.sayHello(userName);
            System.out.println("Total time is " + (System.currentTimeMillis() - start) + " ms ");
            System.out.println("Thrift client result : \n" + result);
        } catch (TTransportException e) {
            e.printStackTrace();
        } catch (TException e) {
            e.printStackTrace();
        } finally {
            if (null != transport) {
                transport.close();
            }
        }
    }

    public static void main(String[] args) {
        TSimpleClient client = new TSimpleClient();
        client.startClient("Nicholas");
    }
}

运行截图

编写TNonblockingServer服务端和客户端测试

TNonblockingServer 使用非阻塞式IO，服务端和客户端需要指定 TFramedTransport 数据传输的方式。

服务端

package com.fengtang.thrift.demo.server;

import com.fengtang.thrift.demo.HelloWorldService;
import com.fengtang.thrift.demo.service.HelloWorldImpl;
import org.apache.thrift.TProcessor;
import org.apache.thrift.protocol.TCompactProtocol;
import org.apache.thrift.server.TNonblockingServer;
import org.apache.thrift.server.TServer;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TNonblockingServerSocket;

/**
 * Create by fengtang
 * 2015/8/14 0014
 * Thrift
 * 使用非阻塞式IO，服务端和客户端需要指定 TFramedTransport 数据传输的方式。
 */
public class TNonbloServer {
    public static final int SERVER_PORT = 8090;

    public void startServer() {
        try {
            System.out.println("HelloWorld TNonblockingServer start ....");

            TProcessor tprocessor = new HelloWorldService.Processor<HelloWorldService.Iface>(
                    new HelloWorldImpl());

            TNonblockingServerSocket tnbSocketTransport = new TNonblockingServerSocket(
                    SERVER_PORT);
            TNonblockingServer.Args tnbArgs = new TNonblockingServer.Args(
                    tnbSocketTransport);
            tnbArgs.processor(tprocessor);
            tnbArgs.transportFactory(new TFramedTransport.Factory());
            tnbArgs.protocolFactory(new TCompactProtocol.Factory());

            // 使用非阻塞式IO，服务端和客户端需要指定TFramedTransport数据传输的方式
            TServer server = new TNonblockingServer(tnbArgs);
            server.serve();
        } catch (Exception e) {
            System.out.println("Server start error!!!");
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        TNonbloServer server = new TNonbloServer();
        server.startServer();
    }
}

客户端

package com.fengtang.thrift.demo.client;

import com.fengtang.thrift.demo.HelloWorldService;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TCompactProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;

/**
 * Create by fengtang
 * 2015/8/14 0014
 * Thrift
 */
public class TNonblockingClient {
    public static final String SERVER_IP = "localhost";
    public static final int SERVER_PORT = 8090;
    public static final int TIMEOUT = 30000;

    /**
     * @param userName
     */
    public void startClient(String userName) {
        TTransport transport = null;
        try {
            transport = new TFramedTransport(new TSocket(SERVER_IP,
                    SERVER_PORT, TIMEOUT));
            // 协议要和服务端一致
            TProtocol protocol = new TCompactProtocol(transport);
            HelloWorldService.Client client = new HelloWorldService.Client(
                    protocol);
            transport.open();
            String result = client.sayHello(userName);
            System.out.println("Thrify client result =: " + result);
        } catch (TTransportException e) {
            e.printStackTrace();
        } catch (TException e) {
            e.printStackTrace();
        } finally {
            if (null != transport) {
                transport.close();
            }
        }
    }

    public static void main(String[] args) {
        TNonblockingClient client = new TNonblockingClient();
        client.startClient("Nicholas");

    }
}

运行截图

编写THsHaServer服务端和客户端测试

THsHaServer 半同步半异步的服务端模型，需要指定为： TFramedTransport 数据传输的方式。

服务端

package com.fengtang.thrift.demo.server;

import com.fengtang.thrift.demo.HelloWorldService;
import com.fengtang.thrift.demo.service.HelloWorldImpl;
import org.apache.thrift.TProcessor;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.server.THsHaServer;
import org.apache.thrift.server.TServer;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TNonblockingServerSocket;

/**
 * Create by fengtang
 * 2015/8/14 0014
 * Thrift
 * 半同步半异步的服务端模型，需要指定为： TFramedTransport 数据传输的方式。
 */
public class ThshaServer {
    public static final int SERVER_PORT = 8090;

    public void startServer() {
        try {
            System.out.println("HelloWorld THsHaServer start ....");

            TProcessor tprocessor = new HelloWorldService.Processor<HelloWorldService.Iface>(
                    new HelloWorldImpl());

            TNonblockingServerSocket tnbSocketTransport = new TNonblockingServerSocket(
                    SERVER_PORT);
            THsHaServer.Args thhsArgs = new THsHaServer.Args(tnbSocketTransport);
            thhsArgs.processor(tprocessor);
            thhsArgs.transportFactory(new TFramedTransport.Factory());
            thhsArgs.protocolFactory(new TBinaryProtocol.Factory());

            //半同步半异步的服务模型
            TServer server = new THsHaServer(thhsArgs);
            server.serve();

        } catch (Exception e) {
            System.out.println("Server start error!!!");
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        ThshaServer server = new ThshaServer();
        server.startServer();
    }
}

客户端

package com.fengtang.thrift.demo.client;

import com.fengtang.thrift.demo.HelloWorldService;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TCompactProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;

/**
 * Create by fengtang
 * 2015/8/14 0014
 * Thrift
 */
public class THsHaClient {
    public static final String SERVER_IP = "localhost";
    public static final int SERVER_PORT = 8090;
    public static final int TIMEOUT = 30000;

    /**
     * @param userName
     */
    public void startClient(String userName) {
        TTransport transport = null;
        try {
            transport = new TFramedTransport(new TSocket(SERVER_IP,
                    SERVER_PORT, TIMEOUT));
            // 协议要和服务端一致
            TProtocol protocol = new TBinaryProtocol(transport);
            HelloWorldService.Client client = new HelloWorldService.Client(protocol);
            transport.open();
            String result = client.sayHello(userName);
            System.out.println("Thrify client result =: " + result);
        } catch (TTransportException e) {
            e.printStackTrace();
        } catch (TException e) {
            e.printStackTrace();
        } finally {
            if (null != transport) {
                transport.close();
            }
        }
    }


    public static void main(String[] args) {
        THsHaClient client = new THsHaClient();
        client.startClient("Nicholas");

    }
}

运行截图

异步客户端调用

服务端

package com.fengtang.thrift.demo.server;

import com.fengtang.thrift.demo.HelloWorldService;
import com.fengtang.thrift.demo.service.HelloWorldImpl;
import org.apache.thrift.TProcessor;
import org.apache.thrift.protocol.TCompactProtocol;
import org.apache.thrift.server.TNonblockingServer;
import org.apache.thrift.server.TServer;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TNonblockingServerSocket;

/**
 * Create by fengtang
 * 2015/8/14 0014
 * Thrift
 */
public class AsynchronousServer {
    public static final int SERVER_PORT = 8090;

    public void startServer() {
        try {
            System.out.println("HelloWorld TNonblockingServer start ....");

            TProcessor tprocessor = new HelloWorldService.Processor<HelloWorldService.Iface>(new HelloWorldImpl());

            TNonblockingServerSocket tnbSocketTransport = new TNonblockingServerSocket(
                    SERVER_PORT);
            TNonblockingServer.Args tnbArgs = new TNonblockingServer.Args(
                    tnbSocketTransport);
            tnbArgs.processor(tprocessor);
            tnbArgs.transportFactory(new TFramedTransport.Factory());
            tnbArgs.protocolFactory(new TCompactProtocol.Factory());

            // 使用非阻塞式IO，服务端和客户端需要指定TFramedTransport数据传输的方式
            TServer server = new TNonblockingServer(tnbArgs);
            server.serve();
        } catch (Exception e) {
            System.out.println("Server start error!!!");
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        AsynchronousServer server = new AsynchronousServer();
        server.startServer();
    }
}

客户端

package com.fengtang.thrift.demo.client;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import com.fengtang.thrift.demo.HelloWorldService;
import org.apache.thrift.TException;
import org.apache.thrift.async.AsyncMethodCallback;
import org.apache.thrift.async.TAsyncClientManager;
import org.apache.thrift.protocol.TCompactProtocol;
import org.apache.thrift.protocol.TProtocolFactory;
import org.apache.thrift.transport.TNonblockingSocket;
import org.apache.thrift.transport.TNonblockingTransport;

import com.fengtang.thrift.demo.HelloWorldService.AsyncClient.sayHello_call;

/**
 * blog http://www.micmiu.com
 *
 * @author Michael
 */
public class AsynchronousClient {

    public static final String SERVER_IP = "localhost";
    public static final int SERVER_PORT = 8090;
    public static final int TIMEOUT = 30000;

    /**
     * @param userName
     */
    public void startClient(String userName) {
        try {
            TAsyncClientManager clientManager = new TAsyncClientManager();
            TNonblockingTransport transport = new TNonblockingSocket(SERVER_IP,
                    SERVER_PORT, TIMEOUT);

            TProtocolFactory tprotocol = new TCompactProtocol.Factory();
            HelloWorldService.AsyncClient asyncClient = new HelloWorldService.AsyncClient(
                    tprotocol, clientManager, transport);
            System.out.println("Client start .....");

            CountDownLatch latch = new CountDownLatch(1);
            AsynCallback callBack = new AsynCallback(latch);
            System.out.println("call method sayHello start ...");
            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 (Exception e) {
            e.printStackTrace();
        }
        System.out.println("startClient end.");
    }

    public class AsynCallback implements AsyncMethodCallback<sayHello_call> {
        private CountDownLatch latch;

        public AsynCallback(CountDownLatch latch) {
            this.latch = latch;
        }

        @Override
        public void onComplete(sayHello_call response) {
            System.out.println("onComplete");
            try {
                System.out.println("AsynCall result =:"
                        + response.getResult().toString());
            } catch (TException e) {
                e.printStackTrace();
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                latch.countDown();
            }
        }

        @Override
        public void onError(Exception exception) {
            System.out.println("onError :" + exception.getMessage());
            latch.countDown();
        }
    }

    /**
     * @param args
     */
    public static void main(String[] args) {
        AsynchronousClient client = new AsynchronousClient();
        client.startClient("Michael");

    }

}

运行截图