区块链基于WebSocket 构建P2P网络

刺骨的言语ヽ痛彻心扉 2023-10-02 20:25 11阅读 0赞

区块链基于WebSocket 构建P2P网络今天我们重点分析，在常见的java web开发中，比如boot框架开发的区块链系统，每个节点既是服务端又是客户端，因此不能引用spring-boot-starter-WebSocket 依赖，而是要引用Java-WebSocket 原生的依赖，以维持不同节点间的长连接。

1、pom引用：

<dependency>
    			<groupId>org.java-websocket</groupId>
    			<artifactId>Java-WebSocket</artifactId>
    			<version>1.3.8</version>
    		</dependency>

2、服务端伪代码

import java.net.InetSocketAddress;
    import java.util.ArrayList;
    import java.util.List;
    
    import javax.annotation.PostConstruct;
    
    import org.java_websocket.WebSocket;
    import org.java_websocket.handshake.ClientHandshake;
    import org.java_websocket.server.WebSocketServer;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.core.annotation.Order;
    import org.springframework.stereotype.Component;
    import org.testng.util.Strings;
    
    /**
     * 基于springboot2.0的websocket服务端
     * 
     * @author nandao
     *
     */
    @Component
    public class P2pPointServer {
      // 日志记录
      private Logger logger = LoggerFactory.getLogger(P2pPointServer.class);
    
      // 本机server的WebSocket端口
      // 多机测试时可改变该值
      private int port = 8001;
    
      // 所有连接到服务端的WebSocket缓存器
      private List<WebSocket> localSockets = new ArrayList<WebSocket>();
    
      public List<WebSocket> getLocalSockets() {
        return localSockets;
      }
    
      public void setLocalSockets(List<WebSocket> localSockets) {
        this.localSockets = localSockets;
      }
    
      /**
       * 初始化P2P Server端
       * @param Server端的端口号port
       */
      @PostConstruct 
      @Order(1)
      public void initServer() {
        /**
         * 初始化WebSocket的服务端 定义内部类对象socketServer，源于WebSocketServer； new
         * InetSocketAddress(port)是WebSocketServer构造器的参数 InetSocketAddress是(IP地址+端口号)类型，亦即端口地址类型
         */
        final WebSocketServer socketServer = new WebSocketServer(new InetSocketAddress(port)) {
          /**
           * 重写5个事件方法，事件发生时触发对应的方法
           */
    
          @Override
          // 创建连接成功时触发
          public void onOpen(WebSocket webSocket, ClientHandshake clientHandshake) {
            sendMessage(webSocket, "服务端开打");
    
            // 当成功创建一个WebSocket连接时，将该链接加入连接池
            localSockets.add(webSocket);
          }
    
          @Override
          // 断开连接时候触发
          public void onClose(WebSocket webSocket, int i, String s, boolean b) {
            logger.info(webSocket.getRemoteSocketAddress() + "客户端与服务器断开连接！");
    
            // 当客户端断开连接时，WebSocket连接池删除该链接
            localSockets.remove(webSocket);
          }
    
          @Override
          // 收到客户端发来消息的时候触发
          public void onMessage(WebSocket webSocket, String msg) {
            logger.info("接收到客户端消息：" + msg);
            sendMessage(webSocket, "收到消息");
          }
    
          @Override
          // 连接发生错误的时候调用,紧接着触发onClose方法
          public void onError(WebSocket webSocket, Exception e) {
            logger.info(webSocket.getRemoteSocketAddress() + "客户端链接错误！");
            localSockets.remove(webSocket);
          }
    
          @Override
          public void onStart() {
            logger.info("WebSocket Server端启动...");
          }
        };
    
        socketServer.start();
        logger.info("监听socketServer端口" + port);
      }
    
      /**
       * 向连接到本机的某客户端发送消息
       * 
       * @param ws
       * @param message
       */
      public void sendMessage(WebSocket ws, String message) {
        logger.info("发送给" + ws.getRemoteSocketAddress().getPort() + "的p2p消息是:" + message);
        ws.send(message);
      }
    
      /**
       * 向所有连接到本机的客户端广播消息
       * 
       * @param message：待广播内容
       */
      public void broatcast(String message) {
        if (localSockets.size() == 0 || Strings.isNullOrEmpty(message)) {
          return;
        }
    
        logger.info("Glad to say broatcast to clients being startted!");
        for (WebSocket socket : localSockets) {
          this.sendMessage(socket, message);
        }
        logger.info("Glad to say broatcast to clients has overred!");
      }
    }

上述伪代码中，服务端调用initServer()方法初始化， 打开8001作为服务端口，其中重写了5个事件方法，事件发生时出发对应的方法，服务启动时调用onStart()，创建成功时触发onOpen()，断开时触发onClose(),收到客户端消息时调onMessage(),收到消息后向客户端返回“收到消息”的消息；连接发生错误时调用的方法onError(),onError调用完毕后触发onClose()方法。

此外，还有给所有连接到本机的客户端广播消息的方法broatcast()。

3、客户端伪代码

import java.net.URI;
    import java.net.URISyntaxException;
    import java.util.ArrayList;
    import java.util.List;
    
    import javax.annotation.PostConstruct;
    
    import org.java_websocket.WebSocket;
    import org.java_websocket.client.WebSocketClient;
    import org.java_websocket.handshake.ServerHandshake;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.core.annotation.Order;
    import org.springframework.stereotype.Component;
    import org.testng.util.Strings;
    
    
    
    /**
     * 基于springboot2.0的websocket客户端
     * 
     * @author nandao
     *
     */
    @Component
    public class P2pPointClient {
      // 日志记录
      private Logger logger = LoggerFactory.getLogger(P2pPointClient.class);
    
      // P2P网络中的节既是Server端，又是Client端。作为Server运行在7001端口（P2pPointServer的port字段），同时作为Client通过ws://localhost:8001连接到服务端
      private String wsUrl = "ws://localhost:8001/";
    
      // 所有客户端WebSocket的连接池缓存
      private List<WebSocket> localSockets = new ArrayList<WebSocket>();
    
    
      public List<WebSocket> getLocalSockets() {
        return localSockets;
      }
    
      public void setLocalSockets(List<WebSocket> localSockets) {
        this.localSockets = localSockets;
      }
    
      /**
       * 连接到服务端
       */
      @PostConstruct  //启动后执行
      @Order(2) //优先级
      public void connectPeer() {
        try {
          // 创建WebSocket的客户端
          final WebSocketClient socketClient = new WebSocketClient(new URI(wsUrl)) {
            @Override
            public void onOpen(ServerHandshake serverHandshake) {
              sendMessage(this, "客户端打开");
              
              localSockets.add(this);
            }
    
            @Override
            public void onMessage(String msg) {
              logger.info("收到服务端发送的消息:" + msg);
            }
    
            @Override
            public void onClose(int i, String msg, boolean b) {
              logger.info("客户端关闭");
              localSockets.remove(this);
            }
    
            @Override
            public void onError(Exception e) {
              logger.info("客户端报错");
              localSockets.remove(this);
            }
          };
          // 客户端 开始连接服务端
          socketClient.connect();
        } catch (URISyntaxException e) {
          logger.info("连接错误:" + e.getMessage());
        }
      }
    
      /**
       * 向服务端发送消息 当前WebSocket的远程Socket地址，就是服务器端
       * 
       * @param ws：
       * @param message
       */
      public void sendMessage(WebSocket ws, String message) {
        logger.info("发送给" + ws.getRemoteSocketAddress().getPort() + "的p2p消息:" + message);
        ws.send(message);
      }
    
      /**
       * 向所有连接过的服务端广播消息
       * 
       * @param message：待广播的消息
       */
      public void broatcast(String message) {
        if (localSockets.size() == 0 || Strings.isNullOrEmpty(message)) {
          return;
        }
    
        logger.info("Glad to say broatcast to servers being startted!");
        for (WebSocket socket : localSockets) {
          this.sendMessage(socket, message);
        }
        logger.info("Glad to say broatcast to servers has overred!");
      }
    }

如上述伪代码，客户端用connectPeer()方法连接到服务端，在方法中客户端通过 ws://localhost:8001 连接到服务器。与服务端类似的地方有，重写五个事件方法。

为了保证connectPeer在服务启动时就能加载，用 @PostConstruct 标记使其加载bean的时候运行，并且只会被服务器执行一次，另外，为了保证服务端先与客户端加载，用 @Order(2) 标识了connectPeer方法。

同时还有，向所有连接过的服务端广播消息的方法broatcast()。

4、p2p网络执行流程如下：

1）当客户端执行connectPeer方法时，成功连接到服务端的onOpen方法时，服务端向客户端发送消息。

2）客户端接到服务端消息后，触发onMessage方法，打印日志；随后客户端onOpen方法执行 sendMessage(....)，发送消息到服务端。

3）服务端接受到客户端消息，触发onMessage，随后调用sendMessage方法。

4）客户端接收到服务端消息，触发onMessage方法..........

总之，后面就是一个你来我往、循环往复的过程。

到此、基于WebSocket 构建P2P网络分享完毕，下篇分享区块链的共识算法，敬请期待！