package tcpunit; import java.io.IOException; import java.io.InputStream; import java.io.PrintWriter; import java.net.InetSocketAddress; import java.net.ServerSocket; import java.net.Socket; import java.util.HashMap; import java.util.Set; public class TCPServer { private final String hostAddress; // 服务器地址 private final int myPort; // 服务器端口号 private ServerSocket myServerSocket; // 服务器ServerSocket private Thread myThread; // 服务器线程 private OnRequestListener requestListener; // 数据请求监听器 private OnResponseListener responseListener; // 数据响应监听 private OnSocketChange socketChangeListener; // 客户端状态改变监听器 private ClientRunner clientRunner; // 客户端处理线程 private Parser parser = new Parser(); private HashMap clients = new HashMap<>(); // 在线客户端列表<客户端编号,客户端对象> public TCPServer(int port) { this(null, port); } public TCPServer(String hostAddress, int port) { this.hostAddress = hostAddress; this.myPort = port; clientRunner = new ClientRunner(); } /** * 安全关闭ServerSocket * * @param serverSocket */ private static final void safeClose(ServerSocket serverSocket) { if (serverSocket != null) { try { serverSocket.close(); } catch (IOException e) { } } } /** * 安全关闭Socket * * @param socket */ private static final void safeClose(Socket socket) { if (socket != null) { try { socket.close(); } catch (IOException e) { } } } /** * 安全关闭BufferedReader * * @param is */ private static void safeClose(InputStream is) { if (is != null) { try { is.close(); } catch (IOException ignored) { } } } /** * 安全关闭PrintWriter * * @param writer */ private static final void safeClose(PrintWriter writer) { writer.close(); } /** * 开始TCP服务 */ public void start() { try { myServerSocket = new ServerSocket(); myServerSocket.bind((hostAddress != null) ? new InetSocketAddress(hostAddress, myPort) : new InetSocketAddress(myPort)); } catch (IOException e) { System.out.println("Port is occupied"); } myThread = new Thread(new Runnable() { int disConnectCount=0; char resData; @Override public void run() { do try { final Socket finalAccept = myServerSocket.accept(); // 监听客户端连接 clientRunner.exec(() -> { InputStream inputStream = null; try { inputStream = finalAccept.getInputStream(); String packet = null; disConnectCount = 0; while (!finalAccept.isClosed()) { do { resData = (char) inputStream.read(); if (resData == 65535) { disConnectCount++; if (disConnectCount > 3) { System.out.println("Connection is disconnected"); throw new IOException(); } } else { disConnectCount = 0; } packet = parser.message_parse_char(resData); } while (packet == null); responseListener.onResponseListener(packet); //System.out.println(packet); } } catch (IOException e) { e.printStackTrace(); System.out.println("Device offline"); safeClose(inputStream); safeClose(finalAccept); removeSocket(clientRunner.id); } }); } catch (IOException e) { e.printStackTrace(); } while (!myServerSocket.isClosed()); } }); myThread.setDaemon(true); myThread.setName("TcpServer Main Listener"); myThread.start(); } private boolean isConnected(Socket socket) { try { socket.sendUrgentData(0xff); return true; } catch (IOException e) { e.printStackTrace(); return false; } } private synchronized void removeSocket(int id) { clients.remove(id); } /** * 停止TCP服务 */ public void stop() { try { safeClose(myServerSocket); if (myThread != null) { myThread.join(); } } catch (Exception e) { e.printStackTrace(); } } /** * 监听服务异常结束 */ public void waitServerStop() { try { myThread.join(); } catch (InterruptedException e) { e.printStackTrace(); } } /** * 客户端处理线程 */ public static class ClientRunner { private long requestCount; public int id; public void exec(Runnable code) { ++requestCount; Thread t = new Thread(code); t.setDaemon(true); t.setName("TcpServer thread (#" + requestCount + ")"); t.start(); } } /** * 监听客户端请求,并可以修改请求内容 */ public interface OnRequestListener { String onRequestListener(String request); } /** * 设置客户端请求监听器 * * @param listener */ public void setOnRequestListener(OnRequestListener listener) { this.requestListener = listener; } /** * 监听网关响应数据,并可以修改请求内容 */ public interface OnResponseListener { String onResponseListener(String response); } /** * 设置网关响应监听器 * * @param listener */ public void setOnResponseListener(OnResponseListener listener) { this.responseListener = listener; } public interface OnSocketChange { void onSocketChange(Set hostAddress); } public void setOnSocketChange(OnSocketChange listener) { this.socketChangeListener = listener; } }