区块链基础：基于websocket的P2P实现

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>cn.hadron</groupId>
    <artifactId>p2p</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <packaging>jar</packaging>

    <name>p2p</name>
    <url>http://maven.apache.org</url>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.java-websocket</groupId>
            <artifactId>Java-WebSocket</artifactId>
            <version>1.3.4</version>
        </dependency>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.9</version>
            <scope>test</scope>
        </dependency>
    </dependencies>
</project>

package cn.hadron.p2p;

import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.List;
import org.java_websocket.WebSocket;
import org.java_websocket.handshake.ClientHandshake;
import org.java_websocket.server.WebSocketServer;

/**
 * p2p服务端
 *
 */
public class P2PServer {

    //所有服务端的WebSocket
    private List<WebSocket> sockets = new ArrayList<WebSocket>();

    public List<WebSocket> getSockets() {
        return sockets;
    }

    /**
     * 初始化P2P Server端
     * @param port :Server端的端口号
     */
    public void initServer(int port) {
        /**
         * 创建有一个WebSocket的服务端
         * 定义了一个WebSocketServer的匿名子类对象socketServer
         * new InetSocketAddress(port)是WebSocketServer构造器的参数
         * InetSocketAddress是(IP地址+端口号)类型，也就是端口地址类型
         */
        final WebSocketServer socketServer = new WebSocketServer(new InetSocketAddress(port)) {
            /**
             * 重写5个事件方法，事件发生时触发该方法
             */

            @Override
            public void onOpen(WebSocket webSocket, ClientHandshake clientHandshake){//创建连接成功时触发
                write(webSocket, "服务端开打");
                //当成功创建一个WebSocket连接时，将该链接加入连接池
                sockets.add(webSocket);
            }
            @Override
            public void onClose(WebSocket webSocket, int i, String s, boolean b) {//断开连接时候触发
                System.out.println(webSocket.getRemoteSocketAddress()+"客户端与服务器断开连接！");
                //当客户端断开连接时，WebSocket连接池删除该链接
                sockets.remove(webSocket);
            }
            @Override
            public void onMessage(WebSocket webSocket, String msg) {//收到客户端发来消息的时候触发
                System.out.println("接收到客户端消息：" + msg);
                write(webSocket, "收到消息");
            }
            @Override
            public void onError(WebSocket webSocket, Exception e) {//连接发生错误的时候调用,紧接着触发onClose方法
                System.out.println(webSocket.getRemoteSocketAddress()+"客户端链接错误！");
                sockets.remove(webSocket);
            }
            @Override
            public void onStart() {
                System.out.println("WebSocket Server端启动...");
            }
        };
        socketServer.start();
        System.out.println("监听socketServer端口" + port);
    }
    /**
     * 向客户端发送消息
     * @param ws
     * @param message
     */
    public void write(WebSocket ws, String message) {
        System.out.println("发送给" + ws.getRemoteSocketAddress().getPort() + "的p2p消息:" + message);
        ws.send(message);
    }
    /**
     * 向所有客户端广播消息
     * @param message
     */
    public void broatcast(String message) {
        if (sockets.size() == 0) {
            return;
        }
        System.out.println("======广播消息开始：");
        for (WebSocket socket : sockets) {
            this.write(socket, message);
        }
        System.out.println("======广播消息结束");
    }

}

package cn.hadron.p2p;

import java.net.URI;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.List;
import org.java_websocket.WebSocket;
import org.java_websocket.client.WebSocketClient;
import org.java_websocket.handshake.ServerHandshake;

/**
 * p2p客户端
 *
 */
public class P2PClient {
    //所有客户端WebSocket的连接池
    private List<WebSocket> sockets = new ArrayList<WebSocket>();

    public List<WebSocket> getSockets() {
        return sockets;
    }

    /**
     * 连接到peer
     */
    public void connectPeer(String peer) { 
        try {
            /**
             * The WebSocketClient is an abstract class that expects a valid "ws://" URI to connect to. 
             * When connected, an instance recieves important events related to the life of the connection. 
             * A subclass must implement onOpen, onClose, and onMessage to be useful. 
             * An instance can send messages to it's connected server via the send method.
             * 
             * 创建有一个WebSocket的客户端
             */
            final WebSocketClient socketClient = new WebSocketClient(new URI(peer)) {
                @Override
                public void onOpen(ServerHandshake serverHandshake) {
                    write(this, "客户端打开");
                    sockets.add(this);
                }

                @Override
                public void onMessage(String msg) {
                    System.out.println("收到服务端发送的消息:" + msg);
                }

                @Override
                public void onClose(int i, String msg, boolean b) {
                    System.out.println("客户端关闭");
                    sockets.remove(this);
                }

                @Override
                public void onError(Exception e) {
                    System.out.println("客户端报错");
                    sockets.remove(this);
                }
            };
            //客户端 开始连接服务端
            socketClient.connect();
        } catch (URISyntaxException e) {
            System.out.println("连接错误:" + e.getMessage());
        }
    }
    /**
     * 向服务端发送消息
     * 当前WebSocket的远程Socket地址，就是服务器端
     * @param ws：
     * @param message
     */
    public void write(WebSocket ws, String message) {
        System.out.println("发送给" + ws.getRemoteSocketAddress().getPort() + "的p2p消息:" + message);
        ws.send(message);
    }
    /**
     * 向所有服务端广播消息
     * @param message
     */
    public void broatcast(String message) {
        if (sockets.size() == 0) {
            return;
        }
        System.out.println("======广播消息开始：");
        for (WebSocket socket : sockets) {
            this.write(socket, message);
        }
        System.out.println("======广播消息结束");
    }
}

package cn.hadron.p2p;

/**
 * P2P网络中每个节点即是服务端又是客户端
 */
public class Main {
    public static void main(String[] args) {
        if(args==null||args.length==0){
            System.out.println("主方法没有参数！");
            return;
        }
        P2PServer p2pServer = new P2PServer();
        P2PClient p2pClient = new P2PClient();
        int p2pPort = Integer.valueOf(args[0]);
        // 启动p2p服务端
        p2pServer.initServer(p2pPort);
        if (args.length == 2 && args[1] != null) {
            // 作为p2p客户端连接p2p服务端
            p2pClient.connectPeer(args[1]);
        }
    }
}

发送给7001的p2p消息:客户端打开
收到服务端发送的消息:服务端开打
收到服务端发送的消息:收到消息

发送给51290的p2p消息:服务端开打
接收到客户端消息：客户端打开
发送给51290的p2p消息:收到消息