Thrift之双向通讯

在实际应用中，却经常会有客户端建立连接后，等待服务端数据的长连接模式，也可以称为双向连接。一、双连接，服务端与客户端都开ThriftServer如果网络环境可控，可以让服务端与客户端互相访问，你可以给服务端与客户端，两者都开一个ThriftServer，也就是两者互为服务端与客户端。这样就可以简单实现互相访问，比如：客户端： <------------------->  服务端：ThriftClient --------------> ThriftServiceThriftService <------------- ThriftClient二、单连接，利用ProcessorFactory中TConnectionInfo的transport定时向客户端发送消息，让thrift保持长连接不立即关闭。thrift是rpc结构的通信框架，rpc结构默认是 【客户端请求 -> 服务端回应 -> 连接断开】 的这种短连接形式，因此rpc默认是没有服务端回调功能，自然也没有长连接。如果要保持连接不关闭且被动接收到对方的数据，需要指定双方连接的service必须为oneway，服务端定时向客户端发送信息（利用客户端发送数据到服务端时连接成功时产生的transport，需客户端也创建服务Processor），同时客户端实时检测transport的状态，以便出现与服务端连接断开的情况出现。具体流程：1、双向连接的service必须为oneway，否则会因为recv函数抛出remote close异常。2、客户端重用建立client的protocol，开线程使用processor.Process(protocol,protocol)监听服务端回调发送过来的消息。3、服务端Processor的创建，使用ProcessorFactory创建Processor，通过getProcessor函数中transport作为向客户端发送消息的client的transport而创建一个Processor。

java实例

定义test.thrift

namespace java com.zychen.thrift service ClientHandshakeService{ oneway void HandShake(); } service ServerCallbackService{ oneway void Push(1: string msg); }

生成接口代码

把thrift-0.9.3.exe和test.thrift文件放在同一个目录。

进入DOS命令执行：thrift-0.9.3.exe --gen java test.thrift

生成文件gen-java/ com/zychen/thrift/Test.java

服务端代码

ClientHandshakeServiceHandler.java

package com.zychen.thrift; import org.apache.thrift.TException; import org.apache.thrift.protocol.TBinaryProtocol; import org.apache.thrift.transport.TTransport; public class ClientHandshakeServiceHandler implements ClientHandshakeService.Iface { public ClientHandshakeServiceHandler(TTransport trans){ client = new ServerCallbackService.Client(new TBinaryProtocol(trans)); } @Override public void HandShake() throws TException { System.out.println("HandShake\n"); StartThread(); } //开始线程 public void StartThread(){ if(threadCallback == null){ stopThread = false; threadCallback = new Thread(new CallbackThread()); threadCallback.start(); } } //停止线程 public void StopThread(){ stopThread = true; if(threadCallback != null){ try { threadCallback.join(); } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); } threadCallback = null; } } @Override protected void finalize() throws Throwable { // TODO Auto-generated method stub StopThread(); super.finalize(); } protected ServerCallbackService.Client client; protected boolean stopThread = false; protected Thread threadCallback = null; class CallbackThread implements Runnable { public void run() { while(true){ if(stopThread){ break; } try { client.Push("aaaaaaa"); Thread.sleep(50); } catch (TException | InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); return; } } } }; } ProcessorFactoryImpl.java

package com.zychen.thrift; import org.apache.thrift.TProcessor; import org.apache.thrift.TProcessorFactory; import org.apache.thrift.transport.TTransport; import com.zychen.thrift.ClientHandshakeService.Processor; public class ProcessorFactoryImpl extends TProcessorFactory { public ProcessorFactoryImpl(TProcessor processor) { super(processor); // TODO Auto-generated constructor stub } @Override public TProcessor getProcessor(TTransport trans) { // TODO Auto-generated method stub //return super.getProcessor(trans); return new ClientHandshakeService.Processor(new ClientHandshakeServiceHandler(trans)); } }ServerTest.java

package com.zychen.thrift; import org.apache.thrift.TProcessorFactory; import org.apache.thrift.protocol.TBinaryProtocol; import org.apache.thrift.protocol.TBinaryProtocol.Factory; import org.apache.thrift.server.TServer; import org.apache.thrift.server.TThreadPoolServer; import org.apache.thrift.server.TThreadPoolServer.Args; import org.apache.thrift.transport.TServerSocket; import org.apache.thrift.transport.TServerTransport; import org.apache.thrift.transport.TTransportException; import com.zychen.thrift.ClientHandshakeService.Processor; public class ServerTest { /** * @param args */ public static void main(String[] args) { TServerSocket tServerSocket; try { tServerSocket = new TServerSocket(9999); TThreadPoolServer.Args targs = new TThreadPoolServer.Args(tServerSocket); TBinaryProtocol.Factory factory = new TBinaryProtocol.Factory(); //获取processFactory TProcessorFactory tProcessorFactory = new ProcessorFactoryImpl(null); targs.protocolFactory(factory); targs.processorFactory(tProcessorFactory); TThreadPoolServer tThreadPoolServer = new TThreadPoolServer(targs); System.out.println("start server..."); tThreadPoolServer.serve(); } catch (TTransportException e) { // TODO Auto-generated catch block e.printStackTrace(); } } } 客户端代码

ServerCallbackServiceImpl.java

package com.zychen.thrift; import java.io.IOException; import org.apache.thrift.TException; import org.apache.thrift.TProcessor; import org.apache.thrift.protocol.TBinaryProtocol; import org.apache.thrift.transport.TSocket; public class ServerCallbackServiceImpl implements ServerCallbackService.Iface{ public ServerCallbackServiceImpl(TSocket socket){ this.socket = socket; } @Override public void Push(String msg) throws TException { // TODO Auto-generated method stub String str = String.format("receive msg %d: %s", nMsgCount++, msg); System.out.println(str); } public void process(){ processor = new ServerCallbackService.Processor<ServerCallbackService.Iface>(this); TBinaryProtocol protocol = new TBinaryProtocol(socket); while (true) { try { //TProcessor，负责调用用户定义的服务接口，从一个接口读入数据，写入一个输出接口 while (processor.process(protocol, protocol)){ //阻塞式方法,不需要内容 System.out.println("走阻塞式方法"); //关闭socket //socket.close(); } //connection lost, return return; }catch (TException e){ System.out.println("连接已断开..."); e.printStackTrace(); return; } } } protected int nMsgCount = 0; protected TSocket socket; protected TProcessor processor; } ClientTest.java

package com.zychen.thrift; import java.io.IOException; import org.apache.thrift.TException; import org.apache.thrift.protocol.TBinaryProtocol; import org.apache.thrift.transport.TSocket; import org.apache.thrift.transport.TTransportException; import com.zychen.thrift.ServerCallbackService.Iface; public class ClientTest { /** * @param args */ public static void main(String[] args) { // TODO Auto-generated method stub TSocket tSocket = new TSocket("localhost",9999); ClientHandshakeService.Client client = new ClientHandshakeService.Client(new TBinaryProtocol(tSocket)); try { tSocket.open(); runMethod(tSocket); //向服务端发送消息 for (int i = 0; i < 100; ++i){ client.HandShake(); Thread.sleep(50); } System.in.read(); tSocket.close(); } catch (TTransportException e) { // TODO Auto-generated catch block e.printStackTrace(); } catch (TException e) { // TODO Auto-generated catch block e.printStackTrace(); } catch (IOException e) { // TODO Auto-generated catch block e.printStackTrace(); } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); } } public static void runMethod(final TSocket tSocket){ Thread thread = new Thread(new Runnable(){ ServerCallbackServiceImpl serverCallbackServiceImpl = new ServerCallbackServiceImpl(tSocket); @Override public void run() { // TODO Auto-generated method stub serverCallbackServiceImpl.process(); } }); thread.start(); }; }

完整代码：下载

参考资料：http://www.cnblogs.com/xiaosuiba/p/4122459.htmlhttp://blog.csdn.net/qq_27989757/article/details/50761051