TTransport
TTransport负责数据的传输，先看类结构图。
阻塞Server使用TServerSocket，它封装了ServerSocket实例，ServerSocket实例监听到客户端的请求会创建一个Socket对象，并将该Socket对象封装为一个TSocket对象用于通信。
非阻塞Server使用TNonblockingServerSocket，它封装了一个ServerSocketChannel实例，ServerSocketChannel实例监听到客户端的请求会创建一个SocketChannel对象，并将该对象封装成一个TNonblockingSocket对象用于之后的通信。当读取完客户端的请求数据后，保存为本地的一个TTransport对象，然后封装为TFramedTransport对象进行处理。
TTransport
TTransport是客户端所有Transport的基类。
public abstract class TTransport { public abstract boolean isOpen();//transport是否打开 public boolean peek() { //是否还有数据要读，如果transport打开时还有数据要读 return isOpen(); } public abstract void open() throws TTransportException;//打开transport读写数据 public abstract void close();//关闭transport //读取len长度的数据到字节数组buf，off表示开始读的位置，返回实际读取的字节数（不一定为len） public abstract int read(byte[] buf, int off, int len) throws TTransportException; //确保读取len长度的数据到字节数组buf，off表示开始读的位置，通过循环调用read()实现，返回实际读取的字节数（len） public int readAll(byte[] buf, int off, int len) throws TTransportException { int got = 0; int ret = 0; while (got < len) { ret = read(buf, off+got, len-got); if (ret <= 0) { throw new TTransportException( "Cannot read. Remote side has closed. Tried to read " + len + " bytes, but only got " + got + " bytes. (This is often indicative of an internal error on the server side. Please check your server logs.)"); } got += ret; } return got; } //将buf中的全部数据写到output public void write(byte[] buf) throws TTransportException { write(buf, 0, buf.length); } //将buf中off位置开始len长度的数据写到output public abstract void write(byte[] buf, int off, int len) throws TTransportException; //清空transport缓存中的数据 public void flush() throws TTransportException {} //获取本地缓存的数据，没有则返回null public byte[] getBuffer() { return null; } //获取本地缓存的下一个读取位置，没有则返回0 public int getBufferPosition() { return 0; } //获取本地缓存的字节数，没有则返回-1 public int getBytesRemainingInBuffer() { return -1; } //从本地缓存中消费n个字节 public void consumeBuffer(int len) {}}
TIOStreamTransport
TIOStreamTransport是面向流的TTransport的子类，阻塞式，实现了流的操作。
public class TIOStreamTransport extends TTransport { private static final Logger LOGGER = LoggerFactory.getLogger(TIOStreamTransport.class.getName()); protected InputStream inputStream_ = null;//输入流 protected OutputStream outputStream_ = null;//输出流 //一波构造函数 protected TIOStreamTransport() {} public TIOStreamTransport(InputStream is) { inputStream_ = is; } public TIOStreamTransport(OutputStream os) { outputStream_ = os; } public TIOStreamTransport(InputStream is, OutputStream os) { inputStream_ = is; outputStream_ = os; } //streams必须在构造时已经被打开，so一直返回true public boolean isOpen() { return true; } //streams必须在构造时已经被打开，不需要这个方法 public void open() throws TTransportException {} //关闭流 public void close() { if (inputStream_ != null) { try { inputStream_.close(); } catch (IOException iox) { LOGGER.warn("Error closing input stream.", iox); } inputStream_ = null; } if (outputStream_ != null) { try { outputStream_.close(); } catch (IOException iox) { LOGGER.warn("Error closing output stream.", iox); } outputStream_ = null; } } //将输入流中的指定数据读取到buf中 public int read(byte[] buf, int off, int len) throws TTransportException { if (inputStream_ == null) { throw new TTransportException(TTransportException.NOT_OPEN, "Cannot read from null inputStream"); } int bytesRead; try { bytesRead = inputStream_.read(buf, off, len); } catch (IOException iox) { throw new TTransportException(TTransportException.UNKNOWN, iox); } if (bytesRead < 0) { throw new TTransportException(TTransportException.END_OF_FILE); } return bytesRead; } //将buf中的数据写出的输出流outputStream_ public void write(byte[] buf, int off, int len) throws TTransportException { if (outputStream_ == null) { throw new TTransportException(TTransportException.NOT_OPEN, "Cannot write to null outputStream"); } try { outputStream_.write(buf, off, len); } catch (IOException iox) { throw new TTransportException(TTransportException.UNKNOWN, iox); } } //清空输出流 public void flush() throws TTransportException { if (outputStream_ == null) { throw new TTransportException(TTransportException.NOT_OPEN, "Cannot flush null outputStream"); } try { outputStream_.flush(); } catch (IOException iox) { throw new TTransportException(TTransportException.UNKNOWN, iox); } }}
TSocket
TSocket类继承自TIOStreamTransport类，实现了对Socket实例的包装。inputStream_和outputStream_通过Socket初始化。
public class TSocket extends TIOStreamTransport { private static final Logger LOGGER = LoggerFactory.getLogger(TSocket.class.getName()); private Socket socket_ = null;//包装socket_ private String host_ = null;//远程host private int port_ = 0;//远程port private int timeout_ = 0;//Socket超时时间 //三个构造函数 public TSocket(Socket socket) throws TTransportException { socket_ = socket; try { socket_.setSoLinger(false, 0); socket_.setTcpNoDelay(true); } catch (SocketException sx) { LOGGER.warn("Could not configure socket.", sx); } if (isOpen()) { try { //初始化inputStream_和outputStream_ inputStream_ = new BufferedInputStream(socket_.getInputStream(), 1024); outputStream_ = new BufferedOutputStream(socket_.getOutputStream(), 1024); } catch (IOException iox) { close(); throw new TTransportException(TTransportException.NOT_OPEN, iox); } } } public TSocket(String host, int port) { this(host, port, 0); } public TSocket(String host, int port, int timeout) { host_ = host; port_ = port; timeout_ = timeout; initSocket(); } //初始化Socket private void initSocket() { socket_ = new Socket(); try { socket_.setSoLinger(false, 0); socket_.setTcpNoDelay(true); socket_.setSoTimeout(timeout_); } catch (SocketException sx) { LOGGER.error("Could not configure socket.", sx); } } public void setTimeout(int timeout) { timeout_ = timeout; try { socket_.setSoTimeout(timeout); } catch (SocketException sx) { LOGGER.warn("Could not set socket timeout.", sx); } } public Socket getSocket() { if (socket_ == null) { initSocket(); } return socket_; } //检查socket_是否连接 public boolean isOpen() { if (socket_ == null) { return false; } return socket_.isConnected(); } //打开socket连接，初始化输入流和输出流 public void open() throws TTransportException { if (isOpen()) { throw new TTransportException(TTransportException.ALREADY_OPEN, "Socket already connected."); } if (host_.length() == 0) { throw new TTransportException(TTransportException.NOT_OPEN, "Cannot open null host."); } if (port_ <= 0) { throw new TTransportException(TTransportException.NOT_OPEN, "Cannot open without port."); } if (socket_ == null) { initSocket(); } try { socket_.connect(new InetSocketAddress(host_, port_), timeout_); inputStream_ = new BufferedInputStream(socket_.getInputStream(), 1024); outputStream_ = new BufferedOutputStream(socket_.getOutputStream(), 1024); } catch (IOException iox) { close(); throw new TTransportException(TTransportException.NOT_OPEN, iox); } } //关闭socket public void close() { super.close(); if (socket_ != null) { try { socket_.close(); } catch (IOException iox) { LOGGER.warn("Could not close socket.", iox); } socket_ = null; } }} TFramedTransport
TFramedTransport作用是通过message之前的4-byte frame size确保读到的message时完整的，防止发生粘包拆包的问题。
//TFramedTransport作用是通过message之前的4-byte frame size确保读到的message时完整的，防止发生粘包拆包的问题public class TFramedTransport extends TTransport { protected static final int DEFAULT_MAX_LENGTH = 16384000;//默认的本地缓存最大字节数 private int maxLength_;//本地缓存最大字节数 private TTransport transport_ = null;//封装的transport_，实际通过该对象实现数据的读取与写入 private final TByteArrayOutputStream writeBuffer_ = new TByteArrayOutputStream(1024);//输出BUffer，将字节数组输出 private TMemoryInputTransport readBuffer_ = new TMemoryInputTransport(new byte[0]);//输入buffer，用于数据读取 //工厂类，用于将一个TTransport实例封装为TFramedTransport实例 public static class Factory extends TTransportFactory { private int maxLength_; public Factory() { maxLength_ = TFramedTransport.DEFAULT_MAX_LENGTH; } public Factory(int maxLength) { maxLength_ = maxLength; } @Override public TTransport getTransport(TTransport base) { return new TFramedTransport(base, maxLength_); } } //两个构造函数 public TFramedTransport(TTransport transport, int maxLength) { transport_ = transport; maxLength_ = maxLength; } public TFramedTransport(TTransport transport) { transport_ = transport; maxLength_ = TFramedTransport.DEFAULT_MAX_LENGTH; } //同transport_的三个方法 public void open() throws TTransportException { transport_.open(); } public boolean isOpen() { return transport_.isOpen(); } public void close() { transport_.close(); } //读数据，一次请求可能调用多次 public int read(byte[] buf, int off, int len) throws TTransportException { if (readBuffer_ != null) { //在一次客户端的请求中第一次调用该方法时，肯定返回got<0，就可以进入 readFrame()方法。 int got = readBuffer_.read(buf, off, len);//readBuffer_已读完或字节数为0时 肯定返回got<0 if (got > 0) { return got; } } readFrame();//从transport_读到本地缓存readBuffer_ return readBuffer_.read(buf, off, len); } @Override public byte[] getBuffer() { return readBuffer_.getBuffer(); } @Override public int getBufferPosition() { return readBuffer_.getBufferPosition(); } @Override public int getBytesRemainingInBuffer() { return readBuffer_.getBytesRemainingInBuffer(); } @Override public void consumeBuffer(int len) { readBuffer_.consumeBuffer(len); } private final byte[] i32buf = new byte[4]; private void readFrame() throws TTransportException { transport_.readAll(i32buf, 0, 4); //读前4个字节，FrameSize int size = decodeFrameSize(i32buf);//由于发送数据方对FrameSize进行了编码，通过解码得到消息的大小 //校验FrameSize是否正确 if (size < 0) { throw new TTransportException("Read a negative frame size (" + size + ")!"); } if (size > maxLength_) { throw new TTransportException("Frame size (" + size + ") larger than max length (" + maxLength_ + ")!"); } byte[] buff = new byte[size]; transport_.readAll(buff, 0, size);//将FrameSize大小的全部数据读到buff readBuffer_.reset(buff);//重置本地缓存 } //write是向本地缓存写入数据，写完后，所有的调用方都要对输出流调用flush进行清空，所以下面一定会进入到flush方法，再通过transport_将本地缓存的数据写出去 public void write(byte[] buf, int off, int len) throws TTransportException { writeBuffer_.write(buf, off, len); } @Override public void flush() throws TTransportException { byte[] buf = writeBuffer_.get(); int len = writeBuffer_.len(); writeBuffer_.reset();//清空 encodeFrameSize(len, i32buf);//对FrameSize进行编码 transport_.write(i32buf, 0, 4);//先写数据大小FrameSize transport_.write(buf, 0, len);//在写真实数据 transport_.flush();//清空 } //以下两个方法是对FrameSize进行编解码，将每个字节高位都位移到低位组成byte数组 public static final void encodeFrameSize(final int frameSize, final byte[] buf) { buf[0] = (byte)(0xff & (frameSize >> 24)); buf[1] = (byte)(0xff & (frameSize >> 16)); buf[2] = (byte)(0xff & (frameSize >> 8)); buf[3] = (byte)(0xff & (frameSize)); } public static final int decodeFrameSize(final byte[] buf) { return ((buf[0] & 0xff) << 24) | ((buf[1] & 0xff) << 16) | ((buf[2] & 0xff) << 8) | ((buf[3] & 0xff)); }} TFramedTransport用到了TMemoryInputTransport类，TMemoryInputTransport封装了一个字节数组byte[]来做输入流的封装。
public final class TMemoryInputTransport extends TTransport { private byte[] buf_;//保存数据的字节数组 private int pos_;//可读数据的开始位置 private int endPos_;//可读数据的的结束位置 public TMemoryInputTransport() { } public TMemoryInputTransport(byte[] buf) { reset(buf); } public TMemoryInputTransport(byte[] buf, int offset, int length) { reset(buf, offset, length); } //重置buf public void reset(byte[] buf) { reset(buf, 0, buf.length); } public void reset(byte[] buf, int offset, int length) { buf_ = buf; pos_ = offset; endPos_ = offset + length; } public void clear() { buf_ = null; } @Override public void close() {} @Override public boolean isOpen() { return true; } @Override public void open() throws TTransportException {} @Override public int read(byte[] buf, int off, int len) throws TTransportException { int bytesRemaining = getBytesRemainingInBuffer();//获取剩余可读的数据大小 int amtToRead = (len > bytesRemaining ? bytesRemaining : len); if (amtToRead > 0) { System.arraycopy(buf_, pos_, buf, off, amtToRead);//将buf_中pos_开始的amtToRead个字节copy到buf中 consumeBuffer(amtToRead);//将可读数据的开始位置增加amtToRead } return amtToRead; } //不支持写 @Override public void write(byte[] buf, int off, int len) throws TTransportException { throw new UnsupportedOperationException("No writing allowed!"); } @Override public byte[] getBuffer() { return buf_; } public int getBufferPosition() { return pos_; } //剩余可读的数据大小 public int getBytesRemainingInBuffer() { return endPos_ - pos_; } //将可读数据的开始位置向后移len public void consumeBuffer(int len) { pos_ += len; }}
TNonblockingTransport
TNonblockingTransport是TTransport的非阻塞抽象子类。
public abstract class TNonblockingTransport extends TTransport { //连接初始化，参考SocketChannel.connect public abstract boolean startConnect() throws IOException; //连接完成，SocketChannel.finishConnect() public abstract boolean finishConnect() throws IOException; //注册到selector public abstract SelectionKey registerSelector(Selector selector, int interests) throws IOException; //读数据到buffer中 public abstract int read(ByteBuffer buffer) throws IOException; //将buffer中的数据写出 public abstract int write(ByteBuffer buffer) throws IOException;}
TNonblockingSocket
TNonblockingSocket是TNonblockingTransport的子类，是非阻塞Socket的实现，用于异步客户端。
public class TNonblockingSocket extends TNonblockingTransport { private static final Logger LOGGER = LoggerFactory.getLogger(TNonblockingSocket.class.getName()); private final SocketAddress socketAddress_;//Host and port info，用于非阻塞连接懒加载 private final SocketChannel socketChannel_;//Java NIO中实现非阻塞读写的核心类 //一波构造函数 public TNonblockingSocket(String host, int port) throws IOException { this(host, port, 0); } public TNonblockingSocket(String host, int port, int timeout) throws IOException { this(SocketChannel.open(), timeout, new InetSocketAddress(host, port)); } public TNonblockingSocket(SocketChannel socketChannel) throws IOException { this(socketChannel, 0, null); if (!socketChannel.isConnected()) throw new IOException("Socket must already be connected"); } private TNonblockingSocket(SocketChannel socketChannel, int timeout, SocketAddress socketAddress) throws IOException { socketChannel_ = socketChannel; socketAddress_ = socketAddress; socketChannel.configureBlocking(false);//设置socketChannel为非阻塞 Socket socket = socketChannel.socket(); socket.setSoLinger(false, 0); socket.setTcpNoDelay(true); setTimeout(timeout); } //将SocketChannel注册到selector上的感兴趣事件，当感兴趣事件就绪时会收到notify public SelectionKey registerSelector(Selector selector, int interests) throws IOException { return socketChannel_.register(selector, interests); } public void setTimeout(int timeout) { try { socketChannel_.socket().setSoTimeout(timeout); } catch (SocketException sx) { LOGGER.warn("Could not set socket timeout.", sx); } } public SocketChannel getSocketChannel() { return socketChannel_; } //检查是否处于连接状态 public boolean isOpen() { // isConnected() does not return false after close(), but isOpen() does return socketChannel_.isOpen() && socketChannel_.isConnected(); } //不要调用该方法，该实现类提供了自己的懒加载方法startConnect()用于打开连接 public void open() throws TTransportException { throw new RuntimeException("open() is not implemented for TNonblockingSocket"); } //读数据到buffer中 public int read(ByteBuffer buffer) throws IOException { return socketChannel_.read(buffer); } //读指定数据到buf中，通过socketChannel_实现 public int read(byte[] buf, int off, int len) throws TTransportException { if ((socketChannel_.validOps() & SelectionKey.OP_READ) != SelectionKey.OP_READ) { throw new TTransportException(TTransportException.NOT_OPEN, "Cannot read from write-only socket channel"); } try { return socketChannel_.read(ByteBuffer.wrap(buf, off, len)); } catch (IOException iox) { throw new TTransportException(TTransportException.UNKNOWN, iox); } } //将buffer中的数据写出 public int write(ByteBuffer buffer) throws IOException { return socketChannel_.write(buffer); } //将buffer中的指定数据写出 public void write(byte[] buf, int off, int len) throws TTransportException { if ((socketChannel_.validOps() & SelectionKey.OP_WRITE) != SelectionKey.OP_WRITE) { throw new TTransportException(TTransportException.NOT_OPEN, "Cannot write to write-only socket channel"); } try { socketChannel_.write(ByteBuffer.wrap(buf, off, len)); } catch (IOException iox) { throw new TTransportException(TTransportException.UNKNOWN, iox); } } //socketChannel_不支持 public void flush() throws TTransportException { } //关闭socket public void close() { try { socketChannel_.close(); } catch (IOException iox) { LOGGER.warn("Could not close socket.", iox); } } //开始初始化 public boolean startConnect() throws IOException { return socketChannel_.connect(socketAddress_); } //是否完成连接 public boolean finishConnect() throws IOException { return socketChannel_.finishConnect(); }} TServerTransport
服务端Transport层共同的父类，主要包括开启监听和接收客户端连接请求两个方法。
public abstract class TServerTransport { //开启监听客户端连接 public abstract void listen() throws TTransportException; //连接请求到达后，创建transport实例 public final TTransport accept() throws TTransportException { TTransport transport = acceptImpl();//具体方法由子类实现 if (transport == null) { throw new TTransportException("accept() may not return NULL"); } return transport; } //关闭监听 public abstract void close(); protected abstract TTransport acceptImpl() throws TTransportException; public void interrupt() {}}
TServerSocket
阻塞服务时使用，TServerSocket对ServerSocket类进行包装，具体实现由TServerSocket完成，代码相对比较简单。
public class TServerSocket extends TServerTransport { private static final Logger LOGGER = LoggerFactory.getLogger(TServerSocket.class.getName()); private ServerSocket serverSocket_ = null;//基于ServerSocket实现 private int clientTimeout_ = 0;//接收Client连接请求的超时时间 public TServerSocket(ServerSocket serverSocket) { this(serverSocket, 0); } //几个构造函数略过。。。创建TServerSocket实例 public TServerSocket(InetSocketAddress bindAddr, int clientTimeout) throws TTransportException { clientTimeout_ = clientTimeout; try { serverSocket_ = new ServerSocket(); serverSocket_.setReuseAddress(true); serverSocket_.bind(bindAddr);// Bind to listening port } catch (IOException ioe) { serverSocket_ = null; throw new TTransportException("Could not create ServerSocket on address " + bindAddr.toString() + "."); } } public void listen() throws TTransportException { if (serverSocket_ != null) { try { serverSocket_.setSoTimeout(0);//等待客户端连接的超时时间，0表示无限超时 } catch (SocketException sx) { LOGGER.error("Could not set socket timeout.", sx); } } } //accept客户端连接并封装为TSocket protected TSocket acceptImpl() throws TTransportException { if (serverSocket_ == null) { throw new TTransportException(TTransportException.NOT_OPEN, "No underlying server socket."); } try { Socket result = serverSocket_.accept(); TSocket result2 = new TSocket(result); result2.setTimeout(clientTimeout_); return result2; } catch (IOException iox) { throw new TTransportException(iox); } } //关闭serverSocket_ public void close() { if (serverSocket_ != null) { try { serverSocket_.close(); } catch (IOException iox) { LOGGER.warn("Could not close server socket.", iox); } serverSocket_ = null; } } public void interrupt() { close(); } public ServerSocket getServerSocket() { return serverSocket_; }}
TNonblockingServerTransport
TNonblockingServerTransport是非阻塞实现的抽象基类，定义了一个向Selector注册对象的抽象方法。
public abstract class TNonblockingServerTransport extends TServerTransport { public abstract void registerSelector(Selector selector);}
TNonblockingServerSocket
TNonblockingServerSocket是TNonblockingServerTransport的具体实现，对ServerSocketChannel的封装。
public class TNonblockingServerSocket extends TNonblockingServerTransport { private static final Logger LOGGER = LoggerFactory.getLogger(TNonblockingServerTransport.class.getName()); private ServerSocketChannel serverSocketChannel = null;//接收Client请求，Java NIO中的Channel private ServerSocket serverSocket_ = null;//serverSocketChannel中的对象 private int clientTimeout_ = 0;//客户端连接建立超时时间 //几个构造函数略过。。。创建TNonblockingServerSocket实例 public TNonblockingServerSocket(InetSocketAddress bindAddr, int clientTimeout) throws TTransportException { clientTimeout_ = clientTimeout; try { serverSocketChannel = ServerSocketChannel.open();//创建serverSocketChannel实例 serverSocketChannel.configureBlocking(false);//设置为非阻塞 serverSocket_ = serverSocketChannel.socket();//创建serverSocket_实例 serverSocket_.setReuseAddress(true); serverSocket_.bind(bindAddr);//绑定监听端口 } catch (IOException ioe) { serverSocket_ = null; throw new TTransportException("Could not create ServerSocket on address " + bindAddr.toString() + "."); } } //开启监听客户端连接 public void listen() throws TTransportException { if (serverSocket_ != null) { try { serverSocket_.setSoTimeout(0);//等待客户端连接的超时时间，0表示无限超时 } catch (SocketException sx) { sx.printStackTrace(); } } } //接收客户端连接的具体实现，接收客户端连接并封装为TNonblockingSocket返回 protected TNonblockingSocket acceptImpl() throws TTransportException { if (serverSocket_ == null) { throw new TTransportException(TTransportException.NOT_OPEN, "No underlying server socket."); } try { SocketChannel socketChannel = serverSocketChannel.accept(); if (socketChannel == null) { return null; } TNonblockingSocket tsocket = new TNonblockingSocket(socketChannel); tsocket.setTimeout(clientTimeout_); return tsocket; } catch (IOException iox) { throw new TTransportException(iox); } } //向selector注册OP_ACCEPT事件，接收新的连接 public void registerSelector(Selector selector) { try { serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT); } catch (ClosedChannelException e) { } } //关闭serverSocket_ public void close() { if (serverSocket_ != null) { try { serverSocket_.close(); } catch (IOException iox) { LOGGER.warn("WARNING: Could not close server socket: " + iox.getMessage()); } serverSocket_ = null; } } public void interrupt() { close(); }}
总结
注意介绍了TTransport的实现方式。
参考资料