springboot socket tcp阻塞协议搭建

Posted 2021-05-15 guxiaohai_

一：简介

1. 网络通信采用三元组：IP地址、端口、协议，Socket是利用三元组解决网络通信的中间件，几乎所有的应用程序都采用Socket通信模型。在自动化控制中最常用到的协议就是TCP协议，因此我们经常会用到基于TCP协议的Socket通信。
   当网络通信时采用TCP协议时，在正式的读写操作之前，服务器与客户端之间必须建立一个连接，当读写操作完成后，双方不再需要这个连接时它们可以释放这个连接，连接的建立是需要三次握手的，而释放则需要四次挥手，因此，每个连接的建立和释放都需要消耗资源和时间。

二：编码

1. 搭建服务

@Slf4j
@Data
@Component
public class TcpSocketServer implements ApplicationRunner {
    @Override
    public void run(ApplicationArguments args){
        init();//开启服务端线程
    }
    private void init(){
        new Thread(new TcpSocketServer.start()).start();
    }

    //开启一个start线程
    private class start implements Runnable{
        private ServerSocket serverSocket;
        // 防止重复创建socket线程链接对象浪费资源
        private ExecutorService executorService = Executors.newCachedThreadPool();

        @SneakyThrows
        @Override
        public void run() {
            try {
                serverSocket =  new ServerSocket(8090);
                log.info("TCP服务已启动，占用端口： {}", serverSocket.getLocalPort());
            } catch (IOException e) {
                log.error("端口冲突,异常信息：{}", e);
                System.exit(0);
            }
            while(true){
                try {
                    // 开启socket监听
                    Socket socket = serverSocket.accept();
                    ClientSocket register = SocketHandler.register(socket);
                    // 在此判断是否重复创建socket对象线程
                    log.info("TCP对象线程： {}",register);
                    if (register != null){
                        executorService.submit(register);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

2. socket任务处理器（封装读写数据，销毁链接等方法）

@Slf4j
public class SocketHandler {

    /**
     * @Author: guwenhai
     * @Description:  将连接的Socket注册到Socket池中
     * @Date: 2020/11/2 11:52
     */
    public static ClientSocket register(Socket socket) {
        ClientSocket clientSocket = new ClientSocket();
        clientSocket.setSocket(socket);
        try {
            clientSocket.setInputStream(new DataInputStream(socket.getInputStream()));
            clientSocket.setOutputStream(new DataOutputStream(socket.getOutputStream()));
            byte[] bytes = new byte[1024];
            clientSocket.getInputStream().read(bytes);
            clientSocket.setKey(new String(bytes, "utf-8"));
            SocketPool.add(clientSocket);
            return clientSocket;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }

    /**
     * @Author: guwenhai
     * @Description:    向指定客户端发送信息
     * @Date: 2020/11/2 11:56
     */
    public static void sendMessage(ClientSocket clientSocket, String message) {
        try {
            clientSocket.getOutputStream().write(HexEcodeUtil.hexStrToBinaryStr(message));
        } catch (IOException e) {
            log.error("发送信息异常：{}", e);
            close(clientSocket);
        }
    }

    /**
     * @Author: guwenhai
     * @Description:    获取指定客户端的上传信息
     * @Date: 2020/11/2 11:56
     */
    public static void onMessage(ClientSocket clientSocket) {
        byte[] keyByte = new byte[1024];
        byte[] msgByte = new byte[1];
        try {
            // 第一次先发送序列号
            if (StringUtils.isEmpty(clientSocket.getKey())) {
                clientSocket.getInputStream().read(keyByte);
                clientSocket.setKey(new String(keyByte, "UTF-8"));
                // 以后发送数据
            } else {
                String info = "";
                while (true) {
                    if (clientSocket.getInputStream().available() > 0) {
                        clientSocket.getInputStream().read(msgByte);
                        String tempStr = HexEcodeUtil.ByteArrayToHexStr(msgByte);
                        info += tempStr;
                        //已经读完
                        if (clientSocket.getInputStream().available() == 0) {
                            //重置,不然每次收到的数据都会累加起来
                            clientSocket.setMessage(info);
                            break;
                        }
                    }
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
            close(clientSocket);
        }
    }

    /**
     * @Author: guwenhai
     * @Description:    指定Socket资源回收
     * @Date: 2020/11/2 11:57
     */
    public static void close(ClientSocket clientSocket) {
        log.info("进行资源回收");
        if (clientSocket != null) {
            log.info("开始回收socket相关资源，其Key为{}", clientSocket.getKey());
            //释放TCP存储
            SocketPool.remove(clientSocket.getKey());
            //停止socket流
            Socket socket = clientSocket.getSocket();
            try {
                socket.shutdownInput();
                socket.shutdownOutput();
            } catch (IOException e) {
                log.error("关闭输入输出流异常，{}", e);
            } finally {
                try {
                    socket.close();
                } catch (IOException e) {
                    log.error("关闭socket异常{}", e);
                }
            }
        }
    }


    /**
     * @Author: guwenhai
     * @Description:    发送数据包，判断数据连接状态
     * @Date: 2020/11/2 11:57
     */
    public static boolean isSocketClosed(ClientSocket clientSocket) {
        try {
            clientSocket.getSocket().sendUrgentData(1);
            return false;
        } catch (IOException e) {
            return true;
        }
    }
}

3. Map存放现以链接服务器的socket

blic class SocketPool {

    private static final ConcurrentHashMap<String, ClientSocket> ONLINE_SOCKET_MAP = new ConcurrentHashMap<>();

    /**
     * @Author: guwenhai
     * @Description:    添加链接池
     * @Date: 2020/11/2 12:01
     */
    public static void add(ClientSocket clientSocket){
        if (clientSocket != null && !clientSocket.getKey().isEmpty()){
            String keyTopic = clientSocket.getKey().replace("\\u0000", "");
            ONLINE_SOCKET_MAP.put(keyTopic, clientSocket);
        }
    }

    /**
     * @Author: guwenhai
     * @Description:    删除链接池
     * @Date: 2020/11/2 12:01
     */
    public static void remove(String key){
        if (StringUtils.isNotBlank(key))
            ONLINE_SOCKET_MAP.remove(key);
    }

    /**
     * @Author: guwenhai
     * @Description:    获取连接池
     * @Date: 2020/11/9 14:10
     */
    public static ClientSocket get(String key){
        if(StringUtils.isNotBlank(key)){
            ClientSocket clientSocket = ONLINE_SOCKET_MAP.get(key);
            return clientSocket;
        }
        return null;
    }
}

4. TCP客户端（监听客户端消息信息）

@Slf4j
@Data
public class ClientSocket implements Runnable {
    //TCP
    private Socket socket;  //TCP客户端连接信息
    private DataInputStream inputStream;    //输入流
    private DataOutputStream outputStream;  //输出流
    private String key; //定义注册包
    private String message; //接收消息
    
    /**
     * @Author: guwenhai
     * @Description: 接收数据
     * @Date: 2020/11/2 13:43
     */
    @Override
    public void run() {
        while (true){
            try {
                SocketHandler.onMessage(this);
                //...this
            } catch (Exception e) {
                e.printStackTrace();
            }
            if (SocketHandler.isSocketClosed(this)){
                log.info("客户端已关闭,其Key值为：{}", this.getKey());
                //关闭对应的服务端资源
                SocketHandler.close(this);
                break;
            }
        }
    }

    /**
     * @Author: guwenhai
     * @Description:    获取TCP连接socket客户端IP
     * @Date: 2021-03-11 14:26
     */
    public static String resolveRemoteIp(InetAddress inetAddress){
        if (inetAddress instanceof Inet4Address) {  //IPv4
            return inetAddress.getHostAddress();
        }else if (inetAddress instanceof Inet6Address) {    //IPv6
            log.info("IPv6:{}",inetAddress);
            return inetAddress.getHostAddress();
        }else { //Not an IP address
            return "";
        }
    }
}