襄阳论坛网站建设,爱站长工具综合查询,如何写网站代码是什么原因,千部小黄油资源百度云1. Socket 配置参数中添加 1.1 读取 Socket 字节时的字节序 1.2 读取数据时#xff0c;单次读取最大缓存值 1.3 从 Socket 读取数据时#xff0c;遵从的数据包结构协议 1.4 服务器返回数据的最大值#xff0c;防止客户端内存溢出 /*** Description: Socket 配置参数*/public…1. Socket 配置参数中添加 1.1 读取 Socket 字节时的字节序 1.2 读取数据时单次读取最大缓存值 1.3 从 Socket 读取数据时遵从的数据包结构协议 1.4 服务器返回数据的最大值防止客户端内存溢出 /*** Description: Socket 配置参数*/public class IOSocketOptions {/*** 是否为调试模式默认为 true*/private static boolean isDebug true;/*** Socket 主机地址*/private SocketAddress socketAddress;/*** Socket 备用主机地址*/private SocketAddress backupAddress;/*** 写入 Socket 字节时的字节序*/private ByteOrder writeOrder;/*** 读取 Socket 字节时的字节序*/private ByteOrder readOrder;/*** 写数据时单个数据包的最大值 默认 100*/private int maxWriteBytes;/*** 读取数据时单次读取最大缓存值默认 50数值越大效率越高但是系统消耗也越大*/private int maxReadBytes;/*** 从 Socket 读取数据时遵从的数据包结构协议在业务层进行定义*/private IMessageProtocol messageProtocol;/*** 服务器返回数据的最大值 (单位Mb) 防止客服端内存溢出*/private int maxResponseDataMb;/*** 连接超时时间(单位毫秒)*/private int connectTimeout;/*** 是否重连 Socket默认为 true*/private boolean isReConnection;/*** Socket 重连管理器*/private AbsReConnection reConnectionManager;/*** Socket 工厂*/private SocketFactory socketFactory;/*** Socket 安全套接字协议相关配置*/private SocketSSLConfig socketSSLConfig;/*** IO 字节流的编码方式默认 UTF-8*/private Charset charsetName;/*** 静态内部类*/public static class Builder {IOSocketOptions socketOptions;// 首先获取一个默认的配置public Builder() {this(getDefaultOptions());}public Builder(IOSocketOptions defaultOptions) {socketOptions defaultOptions;}/*** 设置 Socket 主机地址** param socketAddress* return*/public Builder setSocketAddress(SocketAddress socketAddress) {socketOptions.socketAddress socketAddress;return this;}/*** 设置 Socket 备用主机地址** param backupAddress* return*/public Builder setBackupAddress(SocketAddress backupAddress) {socketOptions.backupAddress backupAddress;return this;}/*** 设置服务器返回数据的允许的最大值单位兆/Mb** param maxResponseDataMb* return*/public Builder setMaxResponseDataMb(int maxResponseDataMb) {socketOptions.maxResponseDataMb maxResponseDataMb;return this;}/*** 设置连接超时时间(单位毫秒)** param connectTimeout* return*/public Builder setConnectTimeout(int connectTimeout) {socketOptions.connectTimeout connectTimeout;return this;}/*** 是否重连 Socket** param reConnection* return*/public Builder setReConnection(boolean reConnection) {socketOptions.isReConnection reConnection;return this;}/*** 设置 Socket 重连管理器** param reConnectionManager* return*/public Builder setReConnectionManager(AbsReConnection reConnectionManager) {socketOptions.reConnectionManager reConnectionManager;return this;}/*** 自定义创建 socket 工厂** param socketFactory*/public Builder setSocketFactory(SocketFactory socketFactory) {socketOptions.socketFactory socketFactory;return this;}/*** 安全套接字协议的配置** param socketSSLConfig* return*/public Builder setSocketSSLConfig(SocketSSLConfig socketSSLConfig) {socketOptions.socketSSLConfig socketSSLConfig;return this;}/*** 设置 Socket 写字节时的字节序** param writeOrder*/public Builder setWriteOrder(ByteOrder writeOrder) {socketOptions.writeOrder writeOrder;return this;}/*** 设置 Socket 读取字节时的字节序** param readOrder* return*/public Builder setReadOrder(ByteOrder readOrder) {socketOptions.readOrder readOrder;return this;}/*** 设置写数据时单个数据包的最大值** param maxWriteBytes* return*/public Builder setMaxWriteBytes(int maxWriteBytes) {socketOptions.maxWriteBytes maxWriteBytes;return this;}/*** 设置读取数据时单次读取最大缓存值** param maxReadBytes* return*/public Builder setMaxReadBytes(int maxReadBytes) {socketOptions.maxReadBytes maxReadBytes;return this;}/*** 设置读取数据的数据结构协议** param messageProtocol*/public Builder setMessageProtocol(IMessageProtocol messageProtocol) {socketOptions.messageProtocol messageProtocol;return this;}/*** 设置 IO 字节流的编码方式默认 UTF-8** param charsetName* return*/public Builder setCharsetName(Charset charsetName) {socketOptions.charsetName charsetName;return this;}public IOSocketOptions build() {return socketOptions;}}/*** 获取默认的配置*/public static IOSocketOptions getDefaultOptions() {IOSocketOptions options new IOSocketOptions();options.socketAddress null;options.backupAddress null;options.isReConnection true;// 是否重连主机默认为 trueoptions.maxWriteBytes 100;options.maxReadBytes 50;options.writeOrder ByteOrder.BIG_ENDIAN;// 大端序options.readOrder ByteOrder.BIG_ENDIAN;// 大端序options.messageProtocol null;// 默认数据包结构协议为 nulloptions.maxResponseDataMb 5;// 默认接收最大值 5MBoptions.reConnectionManager new DefaultReConnection(); // 默认 Socket 重连器options.connectTimeout 10 * 1000; // 连接超时默认 5 秒options.socketFactory null;options.socketSSLConfig null;options.charsetName StandardCharsets.UTF_8;return options;}public Charset getCharsetName() {return charsetName;}public boolean isIsDebug() {return isDebug;}public SocketAddress getSocketAddress() {return socketAddress;}public SocketAddress getBackupAddress() {return backupAddress;}public ByteOrder getWriteOrder() {return writeOrder;}public int getMaxWriteBytes() {return maxWriteBytes;}public ByteOrder getReadOrder() {return readOrder;}public int getMaxReadBytes() {return maxReadBytes;}public IMessageProtocol getMessageProtocol() {return messageProtocol;}public int getMaxResponseDataMb() {return maxResponseDataMb;}public int getConnectTimeout() {return connectTimeout;}public boolean isReConnection() {return isReConnection;}public AbsReConnection getReConnectionManager() {return reConnectionManager;}public SocketFactory getSocketFactory() {return socketFactory;}public SocketSSLConfig getSocketSSLConfig() {return socketSSLConfig;}public static void setIsDebug(boolean isDebug) {IOSocketOptions.isDebug isDebug;}}
2. 定义读取到的数据实体类OriginReadData.java /*** Description: 读取到的数据*/public class OriginReadData implements Serializable {/*** 包头数据*/private byte[] headerData;/*** 包体数据*/private byte[] bodyData;public byte[] getHeaderData() {return headerData;}public void setHeaderData(byte[] headerData) {this.headerData headerData;}public byte[] getBodyData() {return bodyData;}public void setBodyData(byte[] bodyData) {this.bodyData bodyData;}/*** 获取数据体 body 的字符串** return*/public String getBodyString() {//IOSocketOptions.getCharsetNamereturn new String(getBodyData(), StandardCharsets.UTF_8);}/*** 获取完整的数据包括包头和包体** return*/public byte[] getOriginDataBytes() {return Util.concatBytes(getHeaderData(), getBodyData());}}
3. 创建接口 3.1 定义消息数据格式IMessageProtocol.java /*** Description: 消息数据格式*/public interface IMessageProtocol {/*** 获取包头的长度** return*/int getHeaderLength();/*** 获取数据包体的长度根据协议包体长度应该写在包头中在读取数据时用到** param header* param byteOrder* return*/int getBodyLength(byte[] header, ByteOrder byteOrder);/*** 根据协议打包消息格式返回 byte 数组* param sender* return*/byte[] pack(byte[] sender);}3.2 添加 Socket 读取数据行为IOAction.java /*** Description: Socket 读取数据行为*/public interface IOAction {/*** 收到消息的响应*/String ACTION_READ_COMPLETE action_read_complete;} 3.3 添加 Socket 行为监听接口ISocketActionListener.java /*** Description: Socket 行为监听接口*/public interface ISocketActionListener {/*** Socket 连接成功** param socketAddress 主机地址*/void onSocketConnectSuccess(SocketAddress socketAddress);/*** Socket 连接失败** param socketAddress 主机地址* param isNeedReconnect 是否需要重新连接*/void onSocketConnectFail(SocketAddress socketAddress, boolean isNeedReconnect);/*** 断开 Socket 连接** param socketAddress 主机地址* param isNeedReconnect 是否需要重新连接*/void onSocketDisconnect(SocketAddress socketAddress, boolean isNeedReconnect);/*** Socket 数据响应** param socketAddress 主机地址* param originReadData 读取到的数据为类格式*/void onSocketResponse(SocketAddress socketAddress, OriginReadData originReadData);/*** Socket 数据响应** param socketAddress 主机地址* param readData 读取到的数据为 String 类型*/void onSocketResponse(SocketAddress socketAddress, String readData);/*** Socket 数据响应** param socketAddress 主机地址* param readData 读取到的数据为 byte 数组类型*/void onSocketResponse(SocketAddress socketAddress, byte[] readData);}
4. 添加异常类 4.1 可恢复 Socket 读取数据异常ReadRecoverableException.java /*** Description: 可恢复 Socket 读取数据异常*/public class ReadRecoverableException extends Exception {public ReadRecoverableException(String message) {super(message);}} 4.2 不可修复的读取错误异常ReadUnrecoverableException.java /*** Description: 不可修复的读取错误*/public class ReadUnrecoverableException extends Exception {public ReadUnrecoverableException(String message) {super(message);}}
5. 实现 Socket 读取数据 5.1 定义 Socket 读取数据接口 IReader.java /*** Description: Socket 读数据接口*/public interface IReaderT {/*** 打开数据的读取*/void openReader();/*** 读数据*/void read() throws Exception;/*** 关闭数据的读取*/void closeReader();/*** 设置参数** param t*/void setOption(T t);} 5.2 实现 Socket 读取数据接口 /*** Description: Socket 读数据*/public class IOReader implements IReaderIOSocketOptions {/*** 输入流*/private InputStream inputStream;/*** 读取原始数据的缓存空间*/private ByteBuffer originBuffer;/*** Socket 连接管理器*/private IConnectionManager connectionManager;/*** Socket 行为分发器*/private ISocketActionDispatch actionDispatch;/*** Socket 配置参数*/private IOSocketOptions socketOptions;/*** 读数据时余留数据的缓存*/private ByteBuffer remainingBuffer;/*** 读数据线程*/private Thread readerThread;/*** 是否停止线程*/private boolean stopThread;public IOReader(IConnectionManager connectionManager, ISocketActionDispatch actionDispatch) {this.connectionManager connectionManager;this.socketOptions connectionManager.getOptions();this.actionDispatch actionDispatch;}Overridepublic void openReader() {init();if (readerThread null || !readerThread.isAlive()) {readerThread new Thread(readerTask, reader thread);stopThread false;readerThread.start();}}//初始化private void init() {// 获取输入流inputStream connectionManager.getInputStream();// 没有定义消息协议if (socketOptions ! null socketOptions.getMessageProtocol() null) {originBuffer ByteBuffer.allocate(1024 * 4);}}/*** 读取数据任务*/private Runnable readerTask new Runnable() {Overridepublic void run() {while (!stopThread) {try {read();} catch (Exception e) {LogUtil.e(读取数据异常.);e.printStackTrace();// 停止线程stopThread true;release();}}}};Overridepublic void read() throws IOException, ReadRecoverableException, ReadUnrecoverableException {OriginReadData originData new OriginReadData();// 读取数据时数据包结构协议IMessageProtocol messageProtocol this.socketOptions.getMessageProtocol();// 消息协议为 null则直接读取原始消息不建议这样使用因为会发生分包黏包的问题if (messageProtocol null) {readOriginDataFromStream(originData);return;}// 定义了消息协议体// 包头的长度int headerLength messageProtocol.getHeaderLength();// 包头数据的 bufferByteBuffer headerBuffer ByteBuffer.allocate(headerLength);// 设置读取数据时byte 的字节序headerBuffer.order(socketOptions.getReadOrder());// 1.读取 Headerif (remainingBuffer ! null) { // 有余留数据// flip 方法: 一般从 ByteBuffer 读取数据前调用将 limit 设置为当前 position将 position 设置为 0在读取数据时limit 代表可读取数据的有效长度remainingBuffer.flip();// 读取余留数据的长度int length Math.min(remainingBuffer.remaining(), headerLength);// 读取余留的数据headerBuffer.put(remainingBuffer.array(), 0, length);if (length headerLength) { // 余留数据小于一个 Header// 剩余的数据置 nullremainingBuffer null;// 从 stream 中读取剩余的 header 数据readHeaderFromSteam(headerBuffer, headerLength - length);} else {// 移动开始读取数据的指针remainingBuffer.position(headerLength);}} else {// 无余留数据// 从 Stream 读取一个完整的 headercapacity: 容量readHeaderFromSteam(headerBuffer, headerBuffer.capacity());}// 保存 headeroriginData.setHeaderData(headerBuffer.array());// 2.读取 bodyint bodyLength messageProtocol.getBodyLength(originData.getHeaderData(), socketOptions.getReadOrder());if (bodyLength 0) {if (bodyLength socketOptions.getMaxResponseDataMb() * 1024 * 1024) {throw new ReadUnrecoverableException(服务器返回的单次数据超过了规定的最大值可能你的 Socket 消息协议不对一般消息格式为: HeaderBody其中 Header 保存消息长度和类型等Body 保存消息内容请规范好你的协议.);}// 分配包体缓存空间ByteBuffer bodyBuffer ByteBuffer.allocate(bodyLength);bodyBuffer.order(socketOptions.getReadOrder());// 有余留数据if (remainingBuffer ! null) {// 余留数据缓存 包体取数据的起始点int bodyStartPosition remainingBuffer.position();// 获取余留数据长度 与 包体长度 中 取小的值int length Math.min(remainingBuffer.remaining(), bodyLength);// 读 length 大小的余留数据bodyBuffer.put(remainingBuffer.array(), bodyStartPosition, length);// 余留数据缓存移动 positionremainingBuffer.position(bodyStartPosition length);//读取的余留数据刚好等于一个 bodyLengthif (length bodyLength) {if (remainingBuffer.remaining() 0) {// 创建临时数据缓存ByteBuffer temp ByteBuffer.allocate(remainingBuffer.remaining());// 设置字节序temp.order(socketOptions.getReadOrder());// 添加余留数据剩下的数据temp.put(remainingBuffer.array(), remainingBuffer.position(), remainingBuffer.remaining());// 重新赋值余留数据缓存remainingBuffer temp;} else {// 余留数据缓存已取完则当前余留数据缓存置为 nullremainingBuffer null;}// 保存 bodyoriginData.setBodyData(bodyBuffer.array());LogUtil.i(Socket 收到数据: originData.getBodyString());// 分发数据actionDispatch.dispatchAction(IOAction.ACTION_READ_COMPLETE, originData);// 读取结束return;} else {//余留数据缓存已取完当前余留数据缓存置为 nullremainingBuffer null;}}// 无余留或者有部分余留数据则从 Stream 中读取readBodyFromStream(bodyBuffer);// body 保存到 originData 中originData.setBodyData(bodyBuffer.array());} else if (bodyLength 0) { // 没有 body 数据originData.setBodyData(new byte[0]);if (remainingBuffer ! null) { //有余留数据// 处理 body 包体时判断是否有余留数据if (remainingBuffer.hasRemaining()) {//有余留数据// 创建临时数据缓存ByteBuffer temp ByteBuffer.allocate(remainingBuffer.remaining());// 设置字节序temp.order(socketOptions.getReadOrder());// 添加余留数据剩下的数据temp.put(remainingBuffer.array(), remainingBuffer.position(), remainingBuffer.remaining());// 重新赋值余留数据缓存remainingBuffer temp;} else { //余留数据缓存已取完则当前余留数据缓存置为 nullremainingBuffer null;}}} else {//if (bodyLength 0)throw new ReadUnrecoverableException(数据 body 的长度不能小于0);}LogUtil.i(Socket 收到数据: originData.getBodyString());//分发数据actionDispatch.dispatchAction(IOAction.ACTION_READ_COMPLETE, originData);}/*** 读取包头的数据** param headerBuffer 包头的数据缓存* param readLength 包头的长度* throws IOException* throws ReadRecoverableException*/private void readHeaderFromSteam(ByteBuffer headerBuffer, int readLength) throws IOException, ReadRecoverableException {for (int i 0; i readLength; i) {byte[] bytes new byte[1];// 从输入流中读取数据无数据时会阻塞int value inputStream.read(bytes);// -1: 在读取文件中表示读到了末尾在 Socket 中表示一般因为服务器断开了连接if (value -1) {throw new ReadRecoverableException(读取数据失败可能是因为 Socket 跟服务器断开了连接);}//存放包头数据headerBuffer.put(bytes);}}/*** 从 Stream 中读取数据填满 byteBuffer如果读取有多则缓存在 remainingBuffer 中** param byteBuffer 包体的数据缓存* throws IOException* throws ReadRecoverableException*/private void readBodyFromStream(ByteBuffer byteBuffer) throws IOException, ReadRecoverableException {// while 循环直到 byteBuffer 装满数据while (byteBuffer.hasRemaining()) {//从服务器单次读取的最大值byte[] bufferArray new byte[socketOptions.getMaxReadBytes()];int len inputStream.read(bufferArray);if (len -1) {throw new ReadRecoverableException(读取数据失败可能是因为 Socket 跟服务器断开了连接);}//缓存中剩余多少数据int remaining byteBuffer.remaining();// 从 Stream 读到的数据长度 大于 byteBuffer 的剩余空间时if (len remaining) {byteBuffer.put(bufferArray, 0, remaining);// 将多余的数据保存到 remainingBuffer 中缓存等下一次再读取remainingBuffer ByteBuffer.allocate(len - remaining);remainingBuffer.order(socketOptions.getReadOrder());remainingBuffer.put(bufferArray, remaining, len - remaining);} else {// 从 stream 读取的数据小于或者等于 byteBuffer 的剩余空间时byteBuffer.put(bufferArray, 0, len);}}}/*** 当消息协议为 null 时直接返回原始数据** param readData* throws IOException* throws ReadRecoverableException*/private void readOriginDataFromStream(OriginReadData readData) throws IOException, ReadRecoverableException {//用全局 originBuffer 避免重复创建字节数组int len inputStream.read(originBuffer.array());// no more dataif (len -1) {throw new ReadRecoverableException(读取数据失败可能 Socket 跟服务器断开了连接);}//bytes 复制byte[] data new byte[len];originBuffer.get(data, 0, len);readData.setBodyData(data);// Charset.forName(GBK)LogUtil.i(Socket 收到数据: readData.getBodyString());// 分发数据actionDispatch.dispatchAction(IOAction.ACTION_READ_COMPLETE, readData);// 相当于把指针重新指向 position 0originBuffer.clear();}Overridepublic void setOption(IOSocketOptions socketOptions) {this.socketOptions socketOptions;}Overridepublic void closeReader() {try {// 关闭线程释放资源shutDownThread();release();} catch (InterruptedException e) {e.printStackTrace();}}/*** 关闭读取数据线程*/private void shutDownThread() throws InterruptedException {if (readerThread ! null readerThread.isAlive() !readerThread.isInterrupted()) {stopThread true;// 中断线程readerThread.interrupt();readerThread.join();LogUtil.i(shutDownThread);}}/*** 释放资源*/private void release() {if (originBuffer ! null) {originBuffer null;}if (remainingBuffer ! null) {remainingBuffer null;}if (readerThread ! null !readerThread.isAlive()) {readerThread null;}try {if (inputStream ! null) {inputStream.close();}} catch (IOException e) {e.printStackTrace();} finally {inputStream null;}LogUtil.i(release);}}
6. 测试方法 //测试按钮private void initView(View view) {SocketAddress socketAddress new SocketAddress(192.168.1.52, 6688);TcpConnection tcpConnection new TcpConnection(socketAddress);IOSocketOptions socketOptions new IOSocketOptions.Builder().setSocketAddress(socketAddress).build();tcpConnection.setOptions(socketOptions);IOReader reader new IOReader(tcpConnection, null);view.findViewById(R.id.but_connect).setOnClickListener(v - tcpConnection.connect());view.findViewById(R.id.but_start).setOnClickListener(v - reader.openReader());view.findViewById(R.id.but_stop).setOnClickListener(v - reader.closeReader());}
