Android基于socket的TCP通信

package com.frank.tcpclient; import android.os.SystemClock; import android.util.Log; import java.io.IOException; import java.net.InetSocketAddress; import java.nio.ByteBuffer; import java.nio.channels.SelectionKey; import java.nio.channels.Selector; import java.nio.channels.SocketChannel; import java.util.Iterator; import java.util.concurrent.atomic.AtomicBoolean; /** * Created by frank on 2017/4/11. */ public class TcpClient extends Thread implements EventListener { private final static String TAG = TcpClient.class.getSimpleName(); /** * 服务端地址 */ private InetSocketAddress remoteAddress; /** * 客户端地址 */ private InetSocketAddress localAddress; /** * 通信通道 */ private SocketChannel socketChannel; //UDP的是DatagramChannel /** * 选择器 */ private Selector selector; /** * 发送缓冲区 */ private ByteBuffer sendBuffer; /** * 接收缓冲区 */ private ByteBuffer receiveBuffer; /** * 与服务端连接是否断开 */ private AtomicBoolean shutdown = new AtomicBoolean(); /** * 最大缓冲数 */ private static final int MAX_BUFFER_SIZE = 4096; /** * 最大连接次数 */ private static final int MAX_RETRY_COUNT = 10; /** * 连接计数 */ private int count = 0; /** * 连接超时 */ private static final int CONNECT_TIME_OUT = 5000; /** * select时间周期 */ private static final int SELECT_PERIOD = 1000; /** * send时间周期 */ private static final int SEND_PERIOD = 1000; /** * 发送线程 */ private SendThread sendThread; /** * 构造方法，传入服务端地址、客户端地址，分配发送与接收缓冲区，开启发送线程 * @param remoteAddress 服务端地址 * @param localAddress 客户端地址 */ TcpClient(InetSocketAddress remoteAddress, InetSocketAddress localAddress){ this.remoteAddress = remoteAddress; this.localAddress = localAddress; sendBuffer = ByteBuffer.allocate(MAX_BUFFER_SIZE); receiveBuffer = ByteBuffer.allocate(MAX_BUFFER_SIZE); shutdown.set(false); } /** * 开始与服务端建立连接 */ void startConnect(){ if(sendThread != null){ sendThread.interrupt(); } //创建发送线程对象，并且启动线程 sendThread = new SendThread(); sendThread.start(); //打开selector if(selector == null){ try { selector = Selector.open(); } catch (IOException e) { e.printStackTrace(); } } //打开socketChannel if(socketChannel == null){ try { socketChannel = SocketChannel.open(); } catch (IOException e) { e.printStackTrace(); } } try { //设置地址可复用 socketChannel.socket().setReuseAddress(true); //设置连接超时时间 socketChannel.socket().setSoTimeout(CONNECT_TIME_OUT); //设置发送缓冲区大小 socketChannel.socket().setSendBufferSize(MAX_BUFFER_SIZE); //设置接收缓冲区大小 socketChannel.socket().setReceiveBufferSize(MAX_BUFFER_SIZE); //保持长连接 socketChannel.socket().setKeepAlive(true); //Tcp通信不延迟 socketChannel.socket().setTcpNoDelay(true); //设置SelectionKey选项 socketChannel.register(selector, SelectionKey.OP_READ | SelectionKey.OP_WRITE, this); //绑定客户端 socketChannel.socket().bind(localAddress); //连接服务端 socketChannel.connect(remoteAddress); } catch (IOException e) { e.printStackTrace(); } } /** *断开连接 */ void stopConnect(){ //关闭标志位设为true shutdown.set(true); if(socketChannel != null){ try { //关闭输入流 socketChannel.socket().shutdownInput(); //关闭输出流 socketChannel.socket().shutdownOutput(); } catch (IOException e) { e.printStackTrace(); }finally { socketChannel = null; } } if (sendThread == null) return; //停止线程，并且释放该对象 sendThread.interrupt(); sendThread = null; } /** * 判断是否与服务端处于连接状态 * @return 是否与服务端处于连接状态 */ public boolean isConnect(){ return socketChannel.isConnected(); } /** * 向socketChannel写入数据并且发送出去 */ private void sendData(){ if(socketChannel != null && isConnect()){ try { socketChannel.write(sendBuffer); } catch (IOException e) { e.printStackTrace(); } } } /** * 向队列添加数据 * @param data 要发送的数据 */ public void addData(byte data){ sendBuffer.put(data); } /** * 从socketChannel读出数据 * @return 接收到的字节数据 */ private ByteBuffer receiveData(){ ByteBuffer receive = null; if(socketChannel != null && isConnect()){ int length ; int readBuffer = 0; try { //清空缓冲区数据 receiveBuffer.clear(); //循环读取数据 while ((length = socketChannel.read(receiveBuffer)) != 0){ readBuffer += length; } if(readBuffer > 0){ byte[] data = new byte[readBuffer]; //从缓冲区读取数据 receive = receiveBuffer.get(data); } } catch (IOException e) { e.printStackTrace(); }finally { //limit=position;position=0; receiveBuffer.flip(); } } return receive; } @Override public void onRead() { //读取缓冲区数据 ByteBuffer receiveData = receiveData(); if(receiveData == null) return; byte data; while ((data = receiveData.get()) > 0){ Log.i(TAG, "receiveData=" + data); } } @Override public void onConnect() { if(isConnect() && count ++ < MAX_RETRY_COUNT) return; count = 0;//重新计数 } @Override public void onError() { Log.e(TAG, "onError"); } /** * 从selector读取事件，并且回调处理 */ private void select(){ if(selector != null){ //selector读取的事件个数 int n; try { n = selector.select(SELECT_PERIOD); if(n > 0){ Iterator<SelectionKey> iterator = selector.selectedKeys().iterator(); while (iterator.hasNext()){ SelectionKey selectionKey = iterator.next(); iterator.remove(); //检查selectionKey是否有效 if(!selectionKey.isValid()) return; EventListener listener = (EventListener)selectionKey.attachment();