• Java 双工通信与websocket协议

    前言

    最近做项目，需要双工通信，考虑http协议，但是是单向的通信，只能请求响应，不能从服务端推送，如果要服务器推送，方式有很多，http轮训，长轮训，websocket等，实际上tcp传输层是双向通信的，原始的socket就可以实现。现在最常用的是websocket，因为可以复用http的底层tcp连接，方便，当然http3使用udp通信，基于QUIC保证连接可靠。

    Java socket方式

    TCP是面向双向连接的协议，刚好符合条件。server端暴露端口用于连接，实现多方client收发器

    1. package com.feng.socket.admin;
    2.  
    3. import java.io.IOException;
    4. import java.net.ServerSocket;
    5. import java.net.Socket;
    6. import java.util.Map;
    7. import java.util.concurrent.ConcurrentHashMap;
    8. import java.util.concurrent.LinkedBlockingQueue;
    9. import java.util.concurrent.ThreadPoolExecutor;
    10. import java.util.concurrent.TimeUnit;
    11. import java.util.concurrent.atomic.AtomicInteger;
    12.  
    13. public class SocketServer {
    14.     private static AtomicInteger sessionId = new AtomicInteger(10000);
    15.     private static Map<String, Socket> socketMap = new ConcurrentHashMap<>();
    16.  
    17.     private static ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(5, 10, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(20));
    18.     private ServerSocket serverSocket;
    19.  
    20.     public SocketServer(int port) {
    21.         try {
    22.             this.serverSocket = new ServerSocket(port);
    23.         } catch (IOException e) {
    24.             e.printStackTrace();
    25.         }
    26.     }
    27.  
    28.     public void start() {
    29.         //server to client content
    30.         threadPoolExecutor.execute(new WriteRunnable());
    31.         try {
    32.             while (true) {
    33.                 Socket socket = this.serverSocket.accept();
    34.                 String id = sessionId.incrementAndGet()+"";
    35.                 System.out.println("connect from client, the sessionId is :\t"+id);
    36.                 socketMap.put(id, socket);
    37.                 //client to server content
    38.                 threadPoolExecutor.execute(new ReadRunnable(id));
    39.  
    40.             }
    41.         } catch (IOException e) {
    42.             e.printStackTrace();
    43.         }
    44.     }
    45.  
    46.  
    47.     public static Socket getSessionSocket(String sessionId){
    48.         return socketMap.get(sessionId);
    49.     }
    50.     public static void removeSessionSocket(String sessionId){
    51.         socketMap.remove(sessionId);
    52.     }
    53.  
    54.     public static void main(String[] args) {
    55.         new SocketServer(8083).start();
    56.     }
    57. }

    其中收的线程

    1. package com.feng.socket.admin;
    2.  
    3. import java.io.*;
    4. import java.net.Socket;
    5.  
    6. public class ReadRunnable implements Runnable{
    7.  
    8.     private String sessionId;
    9.  
    10.     public ReadRunnable(String sessionId) {
    11.         this.sessionId = sessionId;
    12.     }
    13.  
    14.     @Override
    15.     public void run() {
    16.         try {
    17.             Socket socket = SocketServer.getSessionSocket(sessionId);
    18.             if (socket.isClosed()) {
    19.                 System.out.println("client已断开连接------");
    20.                 return;
    21.             }
    22.             InputStream in = socket.getInputStream();
    23.             Reader reader = new InputStreamReader(in);
    24.             BufferedReader br = new BufferedReader(reader);
    25.  
    26.             while (true) {
    27.                 if (socket.isClosed()) {
    28.                     SocketServer.removeSessionSocket(sessionId);
    29.                     System.out.println("client已断开连接------");
    30.                     break;
    31.                 }
    32.                 try {
    33.                     String line = br.readLine();
    34.                     if (line == null) {
    35.                         System.out.println("client已断开连接------");
    36.                         break;
    37.                     }
    38.                     System.out.println(sessionId + ":" + line);
    39.                     if ("shutdown".equalsIgnoreCase(line)) {
    40.                         socket.close();
    41.                         SocketServer.removeSessionSocket(sessionId);
    42.                         System.out.println("client已断开连接------");
    43.                         break;
    44.                     }
    45.                 } catch (IOException e) {
    46.                     e.printStackTrace();
    47.                 }
    48.             }
    49.             br.close();
    50.             reader.close();
    51.             in.close();
    52.             socket.close();
    53.         } catch (IOException e) {
    54.             throw new RuntimeException(e);
    55.         }
    56.     }
    57. }

     发的方式，使用console输入

    1. package com.feng.socket.admin;
    2.  
    3. import java.io.*;
    4. import java.net.Socket;
    5. import java.util.Map;
    6. import java.util.concurrent.ConcurrentHashMap;
    7.  
    8. public class WriteRunnable implements Runnable{
    9.  
    10.     Map<String, PrintWriter> cache = new ConcurrentHashMap<>();
    11.  
    12.     @Override
    13.     public void run() {
    14.         String line = "";
    15.         PrintWriter pw;
    16.         try (InputStreamReader inputStreamReader = new InputStreamReader(System.in);
    17.              BufferedReader br = new BufferedReader(inputStreamReader)) {
    18.             while (true) {
    19.                 if ((line = br.readLine()) != null) {
    20.                     // 使用 sessionId:xxx 表示发送某个client，发送所有client广播就是迭代发送
    21.                     String sessionId = line.substring(0,5);
    22.                     String data = line.substring(6);
    23.                     Socket socket = SocketServer.getSessionSocket(sessionId);
    24.                     if (socket.isClosed()) {
    25.                         System.out.println("server连接已断开-----");
    26.                     }
    27.                     if (cache.containsKey(sessionId)) {
    28.                         pw = cache.get(sessionId);
    29.                     } else {
    30.                         try {
    31.                             OutputStream out = socket.getOutputStream();
    32.                             pw = new PrintWriter(out);
    33.                             cache.put(sessionId, pw);
    34.                         } catch (IOException e) {
    35.                             throw new RuntimeException(e);
    36.                         }
    37.                     }
    38.                     pw.println(data);
    39.                     pw.flush();
    40.                     if ("shutdown".equalsIgnoreCase(data)) {
    41.                         try {
    42.                             cache.get(sessionId).close();
    43.                             socket.close();
    44.                         } catch (IOException e) {
    45.                             e.printStackTrace();
    46.                         }
    47.                     }
    48.                 }
    49.             }
    50.         } catch (IOException e) {
    51.             throw new RuntimeException(e);
    52.         }
    53.     }
    54. }

    client端代码

    1. package com.feng.socket.client;
    2.  
    3. import java.io.*;
    4. import java.net.Socket;
    5. import java.util.Scanner;
    6.  
    7. public class SocketClient {
    8.     public static void main(String[] args) throws IOException {
    9.         final Socket socket = new Socket("192.168.79.104", 8083);
    10.         final Scanner scanner = new Scanner(System.in);
    11.         new Thread(new Runnable() {
    12.             @Override
    13.             public void run() {
    14.                 String line = "";
    15.                 try (InputStream in = socket.getInputStream();
    16.                      Reader reader = new InputStreamReader(in);
    17.                      BufferedReader br = new BufferedReader(reader)) {
    18.                     while (true) {
    19.                         if (socket.isClosed()) {
    20.                             System.out.println("server断开连接==========");
    21.                             break;
    22.                         }
    23.                         line = br.readLine();
    24.                         if (line != null) {
    25.                             System.out.println(line);
    26.                             if ("shutdown".equalsIgnoreCase(line)) {
    27.                                 break;
    28.                             }
    29.                         } else {
    30.                             System.out.println("server断开连接==========");
    31.                             scanner.close();
    32.                             break;
    33.                         }
    34.                     }
    35.                 } catch (IOException e) {
    36.                     throw new RuntimeException(e);
    37.                 } finally {
    38.                     try {
    39.                         socket.close();
    40.                         System.exit(0);
    41.                     } catch (IOException e) {
    42.                         e.printStackTrace();
    43.                     }
    44.                 }
    45.             }
    46.         }).start();
    47.         new Thread(new Runnable() {
    48.             @Override
    49.             public void run() {
    50.                 String line;
    51.                 try (OutputStream out = socket.getOutputStream();
    52.                      PrintWriter pw = new PrintWriter(out)) {
    53.                     while (true) {
    54.                         if (socket.isClosed()) {
    55.                             System.out.println("client断开连接==========");
    56.                             break;
    57.                         }
    58.                         if (scanner.hasNext()) {
    59.                             line = scanner.next();
    60.                             pw.println(line);
    61.                             pw.flush();
    62.                             out.flush();
    63.                             if ("shutdown".equalsIgnoreCase(line)) {
    64.                                 break;
    65.                             }
    66.                         }
    67.                     }
    68.                     if (!socket.isClosed()) {
    69.                         socket.close();
    70.                     }
    71.                 } catch (IOException e) {
    72.                     throw new RuntimeException(e);
    73.                 }
    74.             }
    75.         }).start();
    76.  
    77.  
    78.     }
    79. }

     相对比较简单，测试如下

    实现了双工，简单模式BIO。

    websocket协议

    实际上websocket就是用来解决这个问题的一个标准，来自于Oracle官网JSR 356, Java API for WebSocket (oracle.com)

     目前的成熟方案是在javaee上实现协议升级，复用Tomcat等的底层容器的TCP连接。spring boot demo，配置bean，当然也可以使用Oracle官网的方式，这里springboot封装了。

    1. package com.feng.socket.admin;
    2.  
    3. import org.springframework.context.annotation.Bean;
    4. import org.springframework.context.annotation.Configuration;
    5. import org.springframework.web.socket.server.standard.ServerEndpointExporter;
    6.  
    7. @Configuration
    8. public class WebSocketConfig {
    9.  
    10.     @Bean
    11.     public ServerEndpointExporter initServerEndpointExporter(){
    12.         return new ServerEndpointExporter();
    13.     }
    14. }

    写一个server端websocket连接收发器

    1. package com.feng.socket.admin;
    2.  
    3. import org.slf4j.Logger;
    4. import org.slf4j.LoggerFactory;
    5. import org.springframework.stereotype.Component;
    6.  
    7. import javax.websocket.*;
    8. import javax.websocket.server.PathParam;
    9. import javax.websocket.server.ServerEndpoint;
    10. import java.io.IOException;
    11. import java.util.Map;
    12. import java.util.concurrent.ConcurrentHashMap;
    13.  
    14. @Component
    15. @ServerEndpoint("/websocket/server/{sessionId}")
    16. public class SocketServer {
    17.  
    18.     private static final Logger LOGGER = LoggerFactory.getLogger(SocketServer.class);
    19.     private static Map<String, Session> sessionMap = new ConcurrentHashMap<>();
    20.  
    21.     private String sessionId = "";
    22.  
    23.     @OnOpen
    24.     public void onOpen(Session session, @PathParam("sessionId") String sessionId) {
    25.         this.sessionId = sessionId;
    26.         sessionMap.put(sessionId, session);
    27.         LOGGER.info("new connect， sessionId is " + sessionId);
    28.     }
    29.  
    30.     @OnClose
    31.     public void onClose() {
    32.         sessionMap.remove(sessionId);
    33.         LOGGER.info("close socket, the sessionId is " + sessionId);
    34.  
    35.     }
    36.  
    37.     @OnMessage
    38.     public void onMessage(String message, Session session) {
    39.         LOGGER.info("--------- receive message: " + message);
    40.     }
    41.  
    42.     @OnError
    43.     public void onError(Session session, Throwable error) {
    44.         LOGGER.error(error.getMessage(), error);
    45.     }
    46.  
    47.     public static void sendMessage(Session session, String message) throws IOException {
    48.         session.getBasicRemote().sendText(message);
    49.     }
    50.  
    51.     public static Session getSession(String sessionId){
    52.         return sessionMap.get(sessionId);
    53.     }
    54. }

    写一个发送接口

    1. package com.feng.socket.admin;
    2.  
    3. import org.springframework.boot.SpringApplication;
    4. import org.springframework.boot.autoconfigure.SpringBootApplication;
    5. import org.springframework.web.bind.annotation.RequestMapping;
    6. import org.springframework.web.bind.annotation.RestController;
    7.  
    8. import javax.websocket.Session;
    9. import java.io.IOException;
    10.  
    11. @RestController
    12. @SpringBootApplication
    13. public class SocketMain {
    14.     public static void main(String[] args) {
    15.         SpringApplication.run(SocketMain.class, args);
    16.     }
    17.  
    18.     @RequestMapping("/msg")
    19.     public String sendMsg(String sessionId, String msg) throws IOException {
    20.         Session session = SocketServer.getSession(sessionId);
    21.         SocketServer.sendMessage(session, msg);
    22.         return "send " + sessionId + " : " + msg;
    23.     }
    24. }

    在使用java-websocket写一个client端

    1.         <dependency>
    2.             <groupId>org.java-websocket</groupId>
    3.             <artifactId>Java-WebSocket</artifactId>
    4.             <version>1.5.3</version>
    5.         </dependency>

    code

    1. package com.feng.socket.client;
    2.  
    3. import org.java_websocket.client.WebSocketClient;
    4. import org.java_websocket.enums.ReadyState;
    5. import org.java_websocket.handshake.ServerHandshake;
    6.  
    7. import java.net.URI;
    8. import java.net.URISyntaxException;
    9. import java.util.Scanner;
    10.  
    11. public class SocketClient {
    12.     public static void main(String[] args) throws URISyntaxException {
    13.         WebSocketClient webSocketClient = new WebSocketClient(new URI("ws://127.0.0.1:8083/websocket/server/10001")) {
    14.             @Override
    15.             public void onOpen(ServerHandshake serverHandshake) {
    16.                 System.out.println(serverHandshake.getHttpStatus() + " : " + serverHandshake.getHttpStatusMessage());
    17.             }
    18.  
    19.             @Override
    20.             public void onMessage(String s) {
    21.                 System.out.println("receive msg is " + s);
    22.             }
    23.  
    24.             @Override
    25.             public void onClose(int i, String s, boolean b) {
    26.                 System.out.println(s);
    27.             }
    28.  
    29.             @Override
    30.             public void onError(Exception e) {
    31.                 e.printStackTrace();
    32.             }
    33.         };
    34.         webSocketClient.connect();
    35.         while (!ReadyState.OPEN.equals(webSocketClient.getReadyState())) {
    36.             System.out.println("wait for connecting ...");
    37.         }
    38.  
    39.         webSocketClient.send("hello");
    40.         Scanner scanner = new Scanner(System.in);
    41.         while (scanner.hasNext()) {
    42.             String line = scanner.next();
    43.             webSocketClient.send(line);
    44.         }
    45.         System.out.println("start websocket client...");
    46.     }
    47. }

    实现服务端与client端双向发送，点对点传输。 

    websocket协议原理

    通过抓包，发现websocket实际上是http协议上的升级

     先client通过http协议发送协商，connection ：Upgrade ，Upgrade的模式是websocket，按照这个理论，我们也可以自定义一种其他Upgrade的协议传输数据。

    协商的key和版本

    然后服务端返回协商的结果

     

    协议达成，如果需要TLS加密传输还要证书传递等

     后面就会ping pong

     

    总结

    这里的双向通信，使用的TCP的特性，websocket也是在http的基础上升级而来，实际上现在的QUIC协议也可以实现在UDP上通信，就是HTTP3.0，就不再适合websocket了，估计会出现新的成熟技术。

  • 相关阅读:
    人人皆知的人工智能真的稳定吗？它的发展前景如何？
    java毕业生设计校园闲置物品交换平台系统计算机源码+系统+mysql+调试部署+lw
    苹果手机照片如何导入电脑？无损快速的传输办法分享！
    通过创建自定义标签来扩展HTML
    EL表达式 Jstl （附上代码理解）
    minikube 快速使用入门 - pod - 外传
    2022年2000元能玩原神的手机推荐 这3款值得买
    react--redux
    英伟达两个最新元宇宙布局
    如何基于YAML设计接口自动化测试框架？看完秒会！
  • 原文地址：https://blog.csdn.net/fenglllle/article/details/125462222