由浅入深了解Thrift（一）——Thrift介绍与用法

图2.2

访问器程序需实现TestThriftService.Iface接口，在实现接口中完成自己要提供的服务，服务器端对服务接口实现的代码如下所示：

package com.test.service;

import org.apache.thrift.TException;

public class TestThriftServiceImpl implements TestThriftService.Iface
{

	@Override
	public String getStr(String srcStr1, String srcStr2) throws TException {
		
		long startTime = System.currentTimeMillis();
		String res = srcStr1 + srcStr2; 
		long stopTime = System.currentTimeMillis();
		
		System.out.println("[getStr]time interval: " + (stopTime-startTime));
		return res;
	}

	@Override
	public int getInt(int val) throws TException {
		long startTime = System.currentTimeMillis();
		int res = val * 10; 
		long stopTime = System.currentTimeMillis();
		
		System.out.println("[getInt]time interval: " + (stopTime-startTime));
		return res;
	}

}

服务器端启动thrift服务框架的程序如下所示，在本例中服务器采用TNonblockingServer工作模式：

package com.test.service;
import org.apache.thrift.TProcessor;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.server.TNonblockingServer;
import org.apache.thrift.server.TServer;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TNonblockingServerSocket;
import org.apache.thrift.transport.TTransportException;
public class testMain {
	private static int m_thriftPort = 12356;
	private static TestThriftServiceImpl m_myService = new TestThriftServiceImpl();
	private static TServer m_server = null;
	private static void createNonblockingServer() throws TTransportException
	{
		TProcessor tProcessor = new TestThriftService.Processor(m_myService);
		TNonblockingServerSocket nioSocket = new TNonblockingServerSocket(m_thriftPort);
		TNonblockingServer.Args tnbArgs = new TNonblockingServer.Args(nioSocket);
		tnbArgs.processor(tProcessor);
		tnbArgs.transportFactory(new TFramedTransport.Factory());
		tnbArgs.protocolFactory(new TBinaryProtocol.Factory());
		// 使用非阻塞式IO，服务端和客户端需要指定TFramedTransport数据传输的方式
		m_server = new TNonblockingServer(tnbArgs);
	}
	public static boolean start()
	{
		try {
			createNonblockingServer();
		} catch (TTransportException e) {
			System.out.println("start server error!" + e);
			return false;
		}
		System.out.println("service at port: " + m_thriftPort);
		m_server.serve();
		return true;
	}
	public static void main(String[] args)
	{
		if(!start())
		{
			System.exit(0);
		}
	}
	
}

代码2.3

在服务器端启动thrift框架的部分代码比较简单，不过在写这些启动代码之前需要先确定服务器采用哪种工作模式对外提供服务，Thrift对外提供几种工作模式，例如：TSimpleServer、TNonblockingServer、TThreadPoolServer、TThreadedSelectorServer等模式，每种服务模式的通信方式不一样，因此在服务启动时使用了那种服务模式，客户端程序也需要采用对应的通信方式。

另外，Thrift支持多种通信协议格式：TCompactProtocol、TBinaryProtocol、TJSONProtocol等，因此，在使用Thrift框架时，客户端程序与服务器端程序所使用的通信协议一定要一致，否则便无法正常通信。

以上述代码2.3采用的TNonblockingServer为例，说明服务器端如何使用Thrift框架，在服务器端创建并启动Thrift服务框架的过程为：

[1]为自己的服务实现类定义一个对象，如代码2.3中的：

TestThriftServiceImplm_myService =newTestThriftServiceImpl();

这里的TestThriftServiceImpl类就是代码2.2中我们自己定义的服务器端对各服务接口的实现类。

[2]定义一个TProcess对象，在根据Thrift文件生成java源码接口文件TestThriftService.java中，Thrift已经自动为我们定义了一个Processor；后续节中将对这个TProcess类的功能进行详细描述；如代码2.3中的：

TProcessor tProcessor = NewTestThriftService.Processor(m_myService);

[3]定义一个TNonblockingServerSocket对象，用于tcp的socket通信，如代码2.3中的：

TNOnblockingServerSocketnioSocket= newTNonblockingServerSocket(m_thriftPort);

在创建server端socket时需要指明监听端口号，即上面的变量：m_thriftPort。

[4]定义TNonblockingServer所需的参数对象TNonblockingServer.Args；并设置所需的参数，如代码2.3中的：

TNonblockingServer.Args tnbArgs = new TNonblockingServer.Args(nioSocket);
tnbArgs.processor(tProcessor);
tnbArgs.transportFactory(new TFramedTransport.Factory());
tnbArgs.protocolFactory(new TBinaryProtocol.Factory());

[5]定义TNonblockingServer对象，并启动该服务，如代码2.3中的：

2.6、  编写客户端代码

Thrift的客户端代码同样需要服务器开头的那两步：添加三个jar包和生成的java接口文件TestThriftService.java。

		 m_transport = new TSocket(THRIFT_HOST, THRIFT_PORT,2000);
		 TProtocol protocol = new TBinaryProtocol(m_transport);
		 TestThriftService.Client testClient = new TestThriftService.Client(protocol);
		
		 try {
			 m_transport.open();
			 
			 String res = testClient.getStr("test1", "test2");
			 System.out.println("res = " + res);
			 m_transport.close();
		} catch (TException e){
				// TODO Auto-generated catch block
				e.printStackTrace();
		}

由代码2.4可以看到编写客户端代码非常简单，只需下面几步即可：

[1]创建一个传输层对象（TTransport），具体采用的传输方式是TFramedTransport，要与服务器端保持一致，即：

m_transport =new TFramedTransport(newTSocket(THRIFT_HOST,THRIFT_PORT, 2000));

这里的THRIFT_HOST, THRIFT_PORT分别是Thrift服务器程序的主机地址和监听端口号，这里的2000是socket的通信超时时间；

[2]创建一个通信协议对象（TProtocol），具体采用的通信协议是二进制协议，这里要与服务器端保持一致，即：

TProtocolprotocol =new TBinaryProtocol(m_transport);

[3]创建一个Thrift客户端对象（TestThriftService.Client），Thrift的客户端类TestThriftService.Client已经在文件TestThriftService.java中，由Thrift编译器自动为我们生成，即：

TestThriftService.ClienttestClient =new TestThriftService.Client(protocol);

[4]打开socket，建立与服务器直接的socket连接，即：

m_transport.open();

[5]通过客户端对象调用服务器服务函数getStr，即：

String res = testClient.getStr("test1","test2");

                System.out.println("res = " +res);

[6]使用完成关闭socket，即：

m_transport.close();

        这里有以下几点需要说明：

[1]在同步方式使用客户端和服务器的时候，socket是被一个函数调用独占的，不能多个调用同时使用一个socket，例如通过m_transport.open()打开一个socket，此时创建多个线程同时进行函数调用，这时就会报错，因为socket在被一个调用占着的时候不能再使用；

[2]可以分时多次使用同一个socket进行多次函数调用，即通过m_transport.open()打开一个socket之后，你可以发起一个调用，在这个次调用完成之后，再继续调用其他函数而不需要再次通过m_transport.open()打开socket；

2.7、  需要注意的问题

（1）Thrift的服务器端和客户端使用的通信方式要一样，否则便无法进行正常通信；

Thrift的服务器端的种模式所使用的通信方式并不一样，因此，服务器端使用哪种通信方式，客户端程序也要使用这种方式，否则就无法进行正常通信了。例如，上面的代码2.3中，服务器端使用的工作模式为TNonblockingServer，在该工作模式下需要采用的传输方式为TFramedTransport，也就是在通信过程中会将tcp的字节流封装成一个个的帧，此时就需要客户端程序也这么做，否则便会通信失败。出现如下问题：

服务器端会爆出如下出错log：

2015-01-06 17:14:52.365 ERROR [Thread-11] [AbstractNonblockingServer.java:348] - Read an invalid frame size of -2147418111. Are you using TFramedTransport on the client side?

客户端则会报出如下出错log：

org.apache.thrift.transport.TTransportException: java.net.SocketException: Connection reset
	at org.apache.thrift.transport.TIOStreamTransport.read(TIOStreamTransport.java:129)
	at org.apache.thrift.transport.TTransport.readAll(TTransport.java:84)
	at org.apache.thrift.protocol.TBinaryProtocol.readAll(TBinaryProtocol.java:362)
	at org.apache.thrift.protocol.TBinaryProtocol.readI32(TBinaryProtocol.java:284)
	at org.apache.thrift.protocol.TBinaryProtocol.readMessageBegin(TBinaryProtocol.java:191)
	at org.apache.thrift.TServiceClient.receiveBase(TServiceClient.java:69)
	at com.browan.freepp.dataproxy.service.DataProxyService$Client.recv_addLiker(DataProxyService.java:877)
	at com.browan.freepp.dataproxy.service.DataProxyService$Client.addLiker(DataProxyService.java:862)
	at com.browan.freepp.dataproxy.service.DataProxyServiceTest.test_Likers(DataProxyServiceTest.java:59)
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.lang.reflect.Method.invoke(Method.java:606)
	at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:47)
	at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
	at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:44)
	at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
	at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:271)
	at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:70)
	at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:50)
	at org.junit.runners.ParentRunner$3.run(ParentRunner.java:238)
	at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:63)
	at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:236)
	at org.junit.runners.ParentRunner.access$000(ParentRunner.java:53)
	at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:229)
	at org.junit.runners.ParentRunner.run(ParentRunner.java:309)
	at org.eclipse.jdt.internal.junit4.runner.JUnit4TestReference.run(JUnit4TestReference.java:50)
	at org.eclipse.jdt.internal.junit.runner.TestExecution.run(TestExecution.java:38)
	at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.runTests(RemoteTestRunner.java:467)
	at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.runTests(RemoteTestRunner.java:683)
	at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.run(RemoteTestRunner.java:390)
	at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.main(RemoteTestRunner.java:197)
Caused by: java.net.SocketException: Connection reset
	at java.net.SocketInputStream.read(SocketInputStream.java:196)
	at java.net.SocketInputStream.read(SocketInputStream.java:122)
	at java.io.BufferedInputStream.fill(BufferedInputStream.java:235)
	at java.io.BufferedInputStream.read1(BufferedInputStream.java:275)
	at java.io.BufferedInputStream.read(BufferedInputStream.java:334)
	at org.apache.thrift.transport.TIOStreamTransport.read(TIOStreamTransport.java:127)
	... 31 more

（2）在服务器端或者客户端直接使用IDL生成的接口文件时，可能会遇到下面两个问题：

[1]、Cannotreduce the visibility of the inherited method fromProcessFunction

[2]、The typeTestThriftService.Processor.getStr must implement theinherited abstract methodProcessFunction.isOneway()

/*String类型返回结果*/
struct ResultStr
{
  1: ThriftResult result,
  2: string value
}

enum ThriftResult
{
  SUCCESS,           /*成功*/
  SERVER_UNWORKING,  /*服务器处于非Working状态*/
  NO_CONTENT,  		 /*请求结果不存在*/
  PARAMETER_ERROR,	 /*参数错误*/
  EXCEPTION,	 	 /*内部出现异常*/
  INDEX_ERROR,		 /*错误的索引或者下标值*/
  UNKNOWN_ERROR, 	 /*未知错误*/
  DATA_NOT_COMPLETE, 	 /*数据不完全*/
  INNER_ERROR, 	 /*内部错误*/
}

namespace java com.browan.freepp.thriftdatatype
const string VERSION = "1.0.1"
/**为ThriftResult添加数据不完全和内部错误两种类型
*/

/****************************************************************************************************
* 定义返回值，
* 枚举类型ThriftResult，表示返回结果，成功或失败，如果失败，还可以表示失败原因
* 每种返回类型都对应一个封装的结构体，该结构体其命名遵循规则："Result" + "具体操作结果类型"，结构体都包含两部分内容：
* 第一部分为枚举类型ThriftResult变量result，表示操作结果,可以 表示成功，或失败，失败时可以给出失败原因
* 第二部分的变量名为value，表示返回结果的内容；
*****************************************************************************************************/
enum ThriftResult
{
  SUCCESS,           /*成功*/
  SERVER_UNWORKING,  /*服务器处于非Working状态*/
  NO_CONTENT,  		 /*请求结果不存在*/
  PARAMETER_ERROR,	 /*参数错误*/
  EXCEPTION,	 	 /*内部出现异常*/
  INDEX_ERROR,		 /*错误的索引或者下标值*/
  UNKNOWN_ERROR 	 /*未知错误*/
  DATA_NOT_COMPLETE 	 /*数据不完全*/
  INNER_ERROR 	 /*内部错误*/
}

/*bool类型返回结果*/
struct ResultBool 
{
  1: ThriftResult result,
  2: bool value
}

/*int类型返回结果*/
struct ResultInt
{
  1: ThriftResult result,
  2: i32 value
}

/*String类型返回结果*/
struct ResultStr
{
  1: ThriftResult result,
  2: string value
}

/*long类型返回结果*/
struct ResultLong
{
  1: ThriftResult result,
  2: i64 value
}



/*double类型返回结果*/
struct ResultDouble
{
  1: ThriftResult result,
  2: double value
}

/*list类型返回结果*/
struct ResultListStr 
{
  1: ThriftResult result,
  2: list value
}

/*Set类型返回结果*/
struct ResultSetStr 
{
  1: ThriftResult result,
  2: set value
}

/*map类型返回结果*/
struct ResultMapStrStr 
{
  1: ThriftResult result,
  2: map value
}

namespace java com.test.service

include "thrift_datatype.thrift"

service TestThriftService
{

	/**
	*value 中存放两个字符串拼接之后的字符串
	*/
	thrift_datatype.ResultStr getStr(1:string srcStr1, 2:string srcStr2),
	
	thrift_datatype.ResultInt getInt(1:i32 val)
	
}