WEB中我们很常见的一种部署方式是在几个tomcat前面加一个nginx做反向代理,此时的nginx有了负载均衡和路由网关的功能。nginx工作在http层,thirft服务工作在tcp层上,所以不能用nginx作为thirft服务的代理(据说nginx可以装一个插件来支持tcp层)。tcp层上的有一个开源的叫HAProxy,用成熟的开源软件有好处,受限制也比较大,本节是用3种方式实现thirft代理,可以更灵活的实现一些自定义功能等。
1 用经典的socket实现:api写起来代码比较简单,就做做转发,但是因为不解析数据协议,且read和accept是阻塞的,所以一个连接会需要两个线程,高并发时,这个缺陷会非常严重。
import java.io.InputStream; import java.io.OutputStream; import java.net.ServerSocket; import java.net.Socket; public class SocketProxy { public static void main(String[] args) throws Exception { ServerSocket serverSocket = null; try { serverSocket = new ServerSocket(4567); System.out.println("Starting the proxy on port 4567..."); while (true) { Socket socket = serverSocket.accept(); //可改成线程池提升性能 new SocketThread(socket).start(); } } catch (Exception e) { e.printStackTrace(); } finally { serverSocket.close(); } } static class SocketThread extends Thread { private Socket socketIn = null; public SocketThread(Socket socket){ socketIn = socket; } public void run() { //可改成缓冲流BufferedInputStream和BufferedOutputStream提升性能 InputStream isIn = null; OutputStream osIn = null; Socket socketOut = null; InputStream isOut = null; OutputStream osOut = null; try { isIn = socketIn.getInputStream(); osIn = socketIn.getOutputStream(); //真正的服务在9090端口 socketOut = new Socket("127.0.0.1", 9090); isOut = socketOut.getInputStream(); osOut = socketOut.getOutputStream(); //可改成线程池提升性能 new StreamThread(isOut, osIn).start(); byte[] buffer = new byte[1024]; int temp = 0; while ((temp = isIn.read(buffer)) != -1) { System.out.println("send:" + new String(buffer, 0, temp, "UTF-8")); osOut.write(buffer, 0, temp); } System.out.println("server end"); } catch (Exception e) { e.printStackTrace(); } finally { closeAll(socketIn, isIn, osIn, socketOut, isOut, osOut); } } private void closeAll(Socket socketIn, InputStream isIn, OutputStream osIn, Socket socketOut, InputStream isOut, OutputStream osOut) { try { isIn.close(); osIn.flush(); osIn.close(); socketIn.close(); isOut.close(); osOut.flush(); osOut.close(); socketOut.close(); } catch (Exception e) { e.printStackTrace(); } } } static class StreamThread extends Thread { private InputStream is = null; private OutputStream os = null; public StreamThread(InputStream inputStream, OutputStream outputStream) { is = inputStream; os = outputStream; } public void run() { byte[] buffer = new byte[1024]; int temp = 0; try { while ((temp = is.read(buffer)) != -1) { System.out.println("receive:" + new String(buffer, 0, temp, "UTF-8")); os.write(buffer, 0, temp); } } catch (Exception e) { e.printStackTrace(); } } } }
2 针对1的缺点,用nio的api实现:api写起来比较恶心,只有一个线程处理一切操作,当然实际应用时当然不会只有一个线程,可以参考thirft的TThreadedSelectorServer是怎么做的,一个线程也非常危险,当一个连接出现问题抛异常的时候,整个程序down掉,其它的连接也一并会被影响。
import java.net.InetSocketAddress; import java.nio.ByteBuffer; import java.nio.channels.SelectionKey; import java.nio.channels.Selector; import java.nio.channels.ServerSocketChannel; import java.nio.channels.SocketChannel; import java.util.Iterator; public class NioProxy { private static final int Listenning_Port = 4567; private static final int Buffer_Size = 1024; private static final int TimeOut = 3000; public static void main(String[] args) throws Exception { // 创建一个在本地端口进行监听的服务Socket信道并设置为非阻塞方式 ServerSocketChannel serverChannel = ServerSocketChannel.open(); serverChannel.socket().bind(new InetSocketAddress(Listenning_Port)); serverChannel.configureBlocking(false); System.out.println("Starting the proxy on port " + Listenning_Port + "..."); // 创建一个选择器并将serverChannel注册到它上面 Selector selector = Selector.open(); serverChannel.register(selector, SelectionKey.OP_ACCEPT); while (true) { // 等待某个信道就绪 if( selector.select(TimeOut) == 0 ){ System.out.println("."); continue; } // 获得就绪信道的键迭代器 Iterator<SelectionKey> keyIter = selector.selectedKeys().iterator(); // 使用迭代器进行遍历就绪信道 while (keyIter.hasNext()) { SelectionKey key = keyIter.next(); // 这种情况是有客户端连接过来,准备一个Channel与之通信,并关联一个真正服务的Channel if (key.isValid() && key.isAcceptable()) { SocketChannel inChannel = ((ServerSocketChannel)key.channel()).accept(); inChannel.configureBlocking(false); SelectionKey icKey = inChannel.register(key.selector(), SelectionKey.OP_READ); SocketChannel outChannel = SocketChannel.open(new InetSocketAddress("127.0.0.1", 9090)); outChannel.configureBlocking(false); SelectionKey ocKey = outChannel.register(selector, SelectionKey.OP_READ); //互相关联 icKey.attach(ocKey); ocKey.attach(icKey); } // 客户端有写入时 if (key.isValid() && key.isReadable()) { // 获得与客户端通信的信道 SocketChannel channel = (SocketChannel)key.channel(); SelectionKey relateKey = (SelectionKey)key.attachment(); SocketChannel relateChannel = (SocketChannel)relateKey.channel(); ByteBuffer buffer = ByteBuffer.allocate(Buffer_Size); buffer.clear(); // 读取信息获得读取的字节数 long bytesRead = channel.read(buffer); if (bytesRead == -1) { // 没有读取到内容的情况 channel.close(); relateChannel.close(); } else { // 将缓冲区准备为数据传出状态 buffer.flip(); relateChannel.write(buffer); // 设置为下一次读取或是写入做准备 key.interestOps(SelectionKey.OP_READ | SelectionKey.OP_WRITE); } } keyIter.remove(); } } } }
3 用thirft的api实现:这个方式可以利用thirft本身的成熟的东西,可以解析数据包,做做认证或者过滤之类的功能。查看thirft的源码,从TSimpleServer的serve方法看进去,我们看到读数据,处理数据,写数据是在processor的process方法里,因此我们可以不用thrift生成的processor,自定义一个万能的processor,接收各种类型的数据再转发出去。
import java.util.ArrayList; import java.util.List; import org.apache.thrift.TApplicationException; import org.apache.thrift.TException; import org.apache.thrift.TProcessor; import org.apache.thrift.protocol.TField; import org.apache.thrift.protocol.TJSONProtocol; import org.apache.thrift.protocol.TList; import org.apache.thrift.protocol.TMap; import org.apache.thrift.protocol.TMessage; import org.apache.thrift.protocol.TMessageType; import org.apache.thrift.protocol.TProtocol; import org.apache.thrift.protocol.TProtocolFactory; import org.apache.thrift.protocol.TProtocolUtil; import org.apache.thrift.protocol.TSet; import org.apache.thrift.protocol.TStruct; import org.apache.thrift.protocol.TType; import org.apache.thrift.server.TServer; import org.apache.thrift.server.TThreadedSelectorServer; import org.apache.thrift.transport.TFastFramedTransport; import org.apache.thrift.transport.TNonblockingServerSocket; import org.apache.thrift.transport.TSocket; import org.apache.thrift.transport.TTransport; import org.apache.thrift.transport.TTransportFactory; public class ThirftProxy { public static void main(String [] args) throws Exception { //自定义的一个processor,非生成代码 ProxyProcessor proxyProcessor = new ProxyProcessor(); //指定的通信协议 TProtocolFactory tProtocolFactory = new TJSONProtocol.Factory(); //指定的通信方式 TTransportFactory tTransportFactory = new TFastFramedTransport.Factory(); int port = 4567; TNonblockingServerSocket tNonblockingServerSocket = new TNonblockingServerSocket(new TNonblockingServerSocket.NonblockingAbstractServerSocketArgs().port(port)); TThreadedSelectorServer.Args tThreadedSelectorServerArgs = new TThreadedSelectorServer.Args(tNonblockingServerSocket); tThreadedSelectorServerArgs.processor(proxyProcessor); tThreadedSelectorServerArgs.protocolFactory(tProtocolFactory); tThreadedSelectorServerArgs.transportFactory(tTransportFactory); //指定的服务器模式 TServer serverEngine = new TThreadedSelectorServer(tThreadedSelectorServerArgs); System.out.println("Starting the proxy on port " + port + "..."); serverEngine.serve(); } static class ProxyProcessor implements TProcessor { @Override public boolean process(TProtocol in, TProtocol out) throws TException { TMessage msg = in.readMessageBegin(); List<ProxyStruct> inDatas = readData(in); in.readMessageEnd(); //转发请求到9090端口 TSocket socket = new TSocket("127.0.0.1", 9090); socket.setTimeout(1000); TTransport transport = socket; transport = new TFastFramedTransport(transport); TProtocol tProtocol = new TJSONProtocol(transport); if ( !transport.isOpen() ) { transport.open(); } tProtocol.writeMessageBegin(msg); wirteData(tProtocol, inDatas); tProtocol.writeMessageEnd(); tProtocol.getTransport().flush(); int seqid = msg.seqid; String methodName = msg.name; msg = tProtocol.readMessageBegin(); if (msg.type == TMessageType.EXCEPTION) { TApplicationException x = TApplicationException.read(tProtocol); tProtocol.readMessageEnd(); throw x; } if (msg.seqid != seqid) { throw new TApplicationException(TApplicationException.BAD_SEQUENCE_ID, methodName + " failed: out of sequence response"); } inDatas = readData(tProtocol); tProtocol.readMessageEnd(); out.writeMessageBegin(msg); wirteData(out, inDatas); out.writeMessageEnd(); out.getTransport().flush(); return true; } private void wirteData(TProtocol out, List<ProxyStruct> outDatas) throws TException { out.writeStructBegin(new TStruct("")); for (ProxyStruct outData : outDatas) { TField field = outData.field; Object value = outData.value; out.writeFieldBegin(field); switch (field.type) { case TType.VOID: break; case TType.BOOL: out.writeBool((boolean)value); break; case TType.BYTE: out.writeByte((byte)value); break; case TType.DOUBLE: out.writeDouble((double)value); break; case TType.I16: out.writeI16((short)value); break; case TType.I32: out.writeI32((int)value); break; case TType.I64: out.writeI64((long)value); break; case TType.STRING: out.writeString((String)value); break; case TType.STRUCT: wirteData(out, (List<ProxyStruct>)value); break; case TType.MAP: //out.writeMapBegin((TMap)value); //out.writeMapEnd(); break; case TType.SET: //out.writeSetBegin((TSet)value); //out.writeSetEnd(); break; case TType.LIST: //out.writeListBegin((TList)value); //out.writeListEnd(); break; case TType.ENUM: break; default: } out.writeFieldEnd(); } out.writeFieldStop(); out.writeStructEnd(); } private List<ProxyStruct> readData(TProtocol in) throws TException { List<ProxyStruct> inDatas = new ArrayList<ProxyStruct>(); TField schemeField; in.readStructBegin(); while (true) { schemeField = in.readFieldBegin(); if (schemeField.type == TType.STOP) { break; } ProxyStruct inData = null; switch (schemeField.type) { case TType.VOID: TProtocolUtil.skip(in, schemeField.type); break; case TType.BOOL: inData = new ProxyStruct(schemeField, in.readBool()); break; case TType.BYTE: inData = new ProxyStruct(schemeField, in.readByte()); break; case TType.DOUBLE: inData = new ProxyStruct(schemeField, in.readDouble()); break; case TType.I16: inData = new ProxyStruct(schemeField, in.readI16()); break; case TType.I32: inData = new ProxyStruct(schemeField, in.readI32()); System.out.println("I32-->" + inData.value); break; case TType.I64: inData = new ProxyStruct(schemeField, in.readI64()); break; case TType.STRING: inData = new ProxyStruct(schemeField, in.readString()); System.out.println("STRING-->" + inData.value); break; case TType.STRUCT: inData = new ProxyStruct(schemeField, readData(in)); break; case TType.MAP: //inData = new ProxyStruct(schemeField, in.readMapBegin()); /** * 这里我懒了,不想写了,readMapBegin返回的TMap对象有3个字段 * keyType,valueType,size,没错就是map的key的类型,value的类型,map的大小 * 从0到size累计循环的按类型读取key和读取value,构造一个hashmap就可以了 */ //in.readMapEnd(); break; case TType.SET: //inData = new ProxyStruct(schemeField, in.readSetBegin()); //同理MAP类型 //in.readSetEnd(); break; case TType.LIST: //inData = new ProxyStruct(schemeField, in.readListBegin()); //同理MAP类型 //in.readListEnd(); break; case TType.ENUM: //Enum类型传输时是个i32 TProtocolUtil.skip(in, schemeField.type); break; default: TProtocolUtil.skip(in, schemeField.type); } if (inData != null ) inDatas.add(inData); in.readFieldEnd(); } in.readStructEnd(); return inDatas; } } //用来存储读取到的各种类型的字段 static class ProxyStruct { public TField field; public Object value; public ProxyStruct(TField tField, Object object) { field = tField; value = object; } } }
注意:这种方式无法支持TTupleProtocol,查看源码知道TTupleProtocol继承了TCompactProtocol,TTupleProtocol编码时为了省空间,没有字段的元信息(id,名字,类型等),只有一个bitset表明哪几个field有值,然后直接用生成的代码读取这些有值的field。TTupleProtocol这种编码方式有个缺点就是服务版本间不兼容。
工程文件结构
依次运行TestServer,SocketProxy/NioProxy/ThirftProxy,TestClient,结果如预期。
推荐一些比较好的相关文章:
通信协议之序列化 http://blog.chinaunix.net/uid-27105712-id-3266286.html
Apache Thrift设计概要 http://calvin1978.blogcn.com/articles/apache-thrift.html
Java NIO浅析 https://zhuanlan.zhihu.com/p/23488863
补充:
jdk1.7后有了aio,和nio有什么不同呢,nio是同步非阻塞的,意思是在真正read数据的时候是阻塞的,只是通过循环信道,事件机制和buffer达到并发的目的,所以对于那些读的数据比较大,读写过程时间长的,nio就不太适合。而aio是异步非阻塞的,read到数据后是通过回调通知相应的方法来继续后面的逻辑,类似于js的callback,所以aio能够胜任那些重量级,读写过程长的任务。写了一个例子AioProxy.java,已经放在附件里了。
java中的AIO https://www.jianshu.com/p/c5e16460047b。
[高并发Java 八] NIO和AIO https://my.oschina.net/hosee/blog/615269。
感谢阅读!附件是eclipse工程源码。
相关推荐
在本篇初学小讲中,我们将重点关注Thrift在负载均衡方面的应用,这对于构建大型分布式系统至关重要。Thrift通过提供一种高效的数据序列化机制以及RPC(远程过程调用)接口,使得不同编程语言之间可以轻松地进行通信...
在本篇小讲中,我们将探讨如何将Thrift与Spring框架结合,以便于构建微服务架构。 首先,让我们了解Thrift的基本工作原理。Thrift IDL(接口定义语言)允许开发者声明服务方法和数据类型,类似于Java中的接口或C++...
在“Apache Thrift 初学小讲(三)【http】”中,我们将深入探讨Thrift如何与HTTP协议相结合,实现基于HTTP的服务通信。 首先,Thrift 提供了一个名为 `ThriftServlet` 的组件,它是将Thrift服务与Java Servlet容器...
Apache Thrift 是一个开源的软件框架,用于构建可伸缩的、跨语言的服务。它结合了接口定义语言(IDL)和库,允许开发者定义数据结构和服务接口,然后自动生成多种编程语言的代码,使得不同语言之间可以高效地进行...
本例改编自Apache Thrift教程: http://mikecvet.wordpress.com/2010/05/13/apache-thrift-tutorial-the-sequel/ http://chanian.com/2010/05/13/thrift-tutorial-a-php-client/ 原教程使用的是c++ server和...
Thrift是一种开源的跨语言服务开发框架,由Facebook于2007年开发并开源,后来成为Apache软件基金会的顶级项目。Thrift的核心思想是通过定义一种中间描述文件(.thrift),来实现数据结构和服务接口的跨语言共享。这...
THRIFT 是一个开源的跨语言服务开发框架,由 Facebook 在 2007 年创建并贡献给了 Apache 基金会。它旨在提供一种高效、可扩展、跨平台的解决方案,用于构建分布式系统中的应用程序接口(API)。通过 THRIFT,开发者...
Thrift是由Facebook开发并在2007年贡献给Apache基金会的一款开源项目。它的主要目的是解决跨平台、跨语言的系统间大数据量传输通信问题。在早期,Facebook面临着系统之间语言环境不一致的问题,因此需要一种能够实现...
在IT领域,尤其是在大数据处理和分布式系统中,HBase是一个重要的NoSQL数据库,它基于Apache Hadoop并提供了高性能、列式存储、可扩展的实时读写能力。而Thrift是一种跨语言的服务框架,由Facebook开发,它允许不同...
Thrift是一种开源的跨语言服务开发框架,由Facebook于2007年开源,现由Apache基金会维护。它允许开发者定义服务接口,然后自动生成多种编程语言的代码,使得不同语言之间可以进行高效、可靠的RPC(远程过程调用)...
Thrift-0.9.1 是一个开源的跨语言服务开发框架,由Facebook于2007年开源,后来成为了Apache软件基金会的顶级项目。这个版本的Thrift源码包包含了一系列的工具和库,使得开发者能够高效地构建分布式系统。Thrift的...
Thrift是一种开源的跨语言服务开发框架,最初由Facebook开发并贡献给Apache基金会。它允许开发者定义服务接口和数据类型,然后自动生成多种编程语言的代码,使得在不同语言之间进行高效、可靠的通信成为可能。在这个...
Thrift是一种跨语言的服务开发框架,最初由Facebook开发,现在是Apache软件基金会的项目。它允许程序员定义服务接口,然后自动生成多种编程语言的代码,使得在这些不同的语言之间进行远程过程调用(RPC)变得容易。...
Thrift和Avro是两种广泛使用的数据序列化和远程过程调用(RPC)框架,它们在分布式系统中扮演着重要角色。在这个实例中,我们将深入理解这两种技术,并探讨它们各自的特性和应用场景。 Thrift是由Facebook开发的一...
Thrift是一种开源的跨语言服务开发框架,由Facebook于2007年开发并开源,后来成为Apache软件基金会的顶级项目。它通过定义一种中间语言(IDL,Interface Definition Language)来描述服务接口,允许开发者在不同的...
Thrift服务开发框架v0.16.0是一款强大的跨语言服务开发工具,它由Facebook开源,现由Apache基金会维护。这个版本的Thrift提供了一种高效、灵活且可扩展的方式来构建分布式系统。Thrift的核心理念是通过定义一种中间...
### Apache Zeppelin 使用指南 #### 一、Apache Zeppelin 概览 Apache Zeppelin 是一款功能强大的基于 Web 的 Notebook 服务器,它为数据科学家提供了一个交互式的环境来探索数据、编写代码并创建可视化报告。...
安装和使用Cassandra-0.8.4,你需要解压下载的"apache-cassandra-0.8.4"压缩包,配置相关的环境变量,启动服务,并使用CQL或者Thrift接口进行数据操作。对于初学者,理解其分布式特性和数据模型是入门的关键。同时,...