electricEnergy/src/tcpunit/TCPServer.java

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<Integer, Socket> 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<String> hostAddress);
    }

    public void setOnSocketChange(OnSocketChange listener) {
        this.socketChangeListener = listener;
    }

}