Apache Thrift 初学小讲（五）【代理】
程序员文章站
2022-05-04 11:21:48
...
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工程源码。
- ThriftTest.rar (2 MB)
- 下载次数: 4
上一篇： 敏捷体验设计师应该具备的12项技能