一、pom依赖

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

二、服务端代码

package com.peck.blockchain.p2p;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;import javax.annotation.PostConstruct;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.List;/*** @author songanning* @ClassName:P2pPointServer* @Description:基于springBoot的WebSocket服务端* @Date:2021/11/23 10:18*/
@Component
public class P2pPointServer {/*** 日志记录*/private Logger logger = LoggerFactory.getLogger(P2pPointServer.class);/*** 本机Server的WebSocket端口* 多机测试可以改变该值*/private int port = 7001;/*** 所有连接到服务端的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端**/@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);}@Overridepublic void onStart() {logger.info("北京服务端监听socketServer端启动......");}};socketServer.start();logger.info("北京服务端监听socketServer端口：" + port);}/*** 向连接到本机的某客户端发送消息*/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!");}
}

三、客户端代码

package com.peck.blockchain.p2p;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;import javax.annotation.PostConstruct;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.List;/*** @author songanning* @ClassName:P2pPointClient* @Description:基于SpringBoot的WebSocket客户端* @Date:2021/11/23 11:09*/
@Component
public class P2pPointClient {/*** 日志记录*/private Logger logger = LoggerFactory.getLogger(P2pPointClient.class);/*** p2p 网络中的节点既是服务端，又是客户端。作为服务端运行在7001端口* (P2pPointServer 的port字段)，同时作为客户通过ws://localhost:7001连接到服务器*/private String wsUrl = "ws://localhost:7001/";/*** 所有客户端WebSocket的连接池缓存*/private List<WebSocket> localSockets = new ArrayList<>();public List<WebSocket> getLocalSockets() {return localSockets;}public void setLocalSockets(List<WebSocket> localSockets) {this.localSockets = localSockets;}/*** 连接到服务器*/@PostConstruct@Order(2)public void connentPeer() {try {final WebSocketClient socketClient = new WebSocketClient(new URI(wsUrl)) {@Overridepublic void onOpen(ServerHandshake serverHandshake) {sendMessage(this,"北京客户端成功创建客户端");localSockets.add(this);}@Overridepublic void onMessage(String msg) {logger.info("北京客户端收到北京服务端发送的消息：" + msg);}@Overridepublic void onClose(int i, String s, boolean b) {logger.info("北京客户端关闭");localSockets.remove(this);}@Overridepublic void onError(Exception e) {logger.info("北京客户端报错");localSockets.remove(this);}};//客户端开始连接服务端socketClient.connect();}catch (URISyntaxException e) {e.printStackTrace();}}/*** 向服务端发送消息，当前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 broadcast(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!");}
}