Java语言实现区块链（五）

一、集成WebSocket，实现P2P网络通信

WebSocket官方文档：https://github.com/TooTallNate/Java-WebSocket

（1）引入坐标。

compile "org.java-websocket:Java-WebSocket:1.3.8"

（2）在websocket包创建两个类，一个代表websocket客户端，一个代表websocket服务端。

public class MyClient extends WebSocketClient {
    // 客户端的名称
    private String name;

    // 在构造函数中传入连接服务端的地址，并指定客户端的名称
    public MyClient(URI serverUri, String name) {
        super(serverUri);
        this.name = name;
    }

    @Override
    public void onOpen(ServerHandshake handshakedata) {
        System.out.println("客户端" + name + "打开了连接...");
    }

    @Override
    public void onMessage(String message) {
        System.out.println("客户端" + name + "收到服务端发送过来的消息...");
    }

    @Override
    public void onClose(int code, String reason, boolean remote) {
        System.out.println("客户端" + name + "关闭了连接...");
    }

    @Override
    public void onError(Exception ex) {
        System.out.println("客户端" + name + "发生错误...");
    }
}

public class MyServer extends WebSocketServer {
    private int port;

    // 在构造函数中指定监听的端口号
    public MyServer(int port) {
        super(new InetSocketAddress(port));
        this.port = port;
    }

    @Override
    public void onOpen(WebSocket conn, ClientHandshake handshake) {
        System.out.println("WebSocket_" + port + "打开了连接...");
    }

    @Override
    public void onClose(WebSocket conn, int code, String reason, boolean remote) {
        System.out.println("WebSocket_" + port + "关闭了连接...");
    }

    @Override
    public void onMessage(WebSocket conn, String message) {
        System.out.println("WebSocket_" + port + "接收到消息...");
    }

    @Override
    public void onError(WebSocket conn, Exception ex) {
        System.out.println("WebSocket_" + port + "发送了错误...");
    }

    @Override
    public void onStart() {
        System.out.println("WebSocket_" + port + "启动成功...");
    }

    // 启动服务端
    public void startServer() {
        // 因为WebServerSocket继承了Runnable类，所以可以启动线程去执行它
        new Thread(this).start();
    }
}

（3）修改BlockChainApplication类，让springboot启动时候可以动态指定端口号。

@SpringBootApplication
public class BlockchainqsApplication {
    public static int port;

    public static void main(String[] args) {
        System.out.println("请输入端口号：");
        Scanner sc = new Scanner(System.in);
        port = sc.nextInt();
        SpringApplicationBuilder application = new SpringApplicationBuilder(BlockchainqsApplication.class)
                .properties("server.port=" + port);
        application.run(args);
    }

}

（4）在BlockController中定义以下几个方法：

init()：启动websocket服务，端口号为springboot端口号加1。

regist()：注册节点

conn()：连接节点

broadcast()：广播消息

// 记录已经注册了的节点（这里是端口号）
private HashSet nodes = new  HashSet<>();
private MyServer myServer;

// 初始化myServer
@PostConstruct
public void init() {
    // 启动服务端，端口号为springboot端口号加1
    myServer = new MyServer(BlockchainqsApplication.port + 1);
    myServer.startServer();
}

// 注册节点
@RequestMapping(value = "/regist")
public String regist(String node) {
    nodes.add(node);
    return "注册成功";
}

// 连接
@RequestMapping(value = "/conn")
public String conn() {
    try {
        // 遍历所有已经注册了的节点，然后创建对应客户端去连接这些节点
        // 相当于每个节点都对应着一个MyClient对象
        for (String node : nodes) {
            URI uri = new URI("ws://localhost:"+ node);
            MyClient myClient = new MyClient(uri, node);
            myClient.connect();
        }
        return "连接成功";
    } catch (URISyntaxException e) {
        return "连接失败：" + e.getMessage();
    }
 }

 // 广播消息
 @RequestMapping(value = "/broadcast")
 public String broadcast(String msg) {
     // 广播消息，其实就是向已注册该WebSocket服务的客户端发送消息
     myServer.broadcast(msg);
     return "广播成功";
 }

（5）修改页面，指定按钮组。

    注册节点
    连接节点
    广播

（6）定义事件函数。

// 注册节点
function regist() {
    // 获取用户输入的内容
    var node = $("#node").val();
    // 显示进度条
    loading.baosight.showPageLoadingMsg(false);
    // 发起请求
    $.post("regist", "node=" + node, function (data) {
        // 展示操作结果
        $("#result").html(data)
        // 清空输入框
        $("#node").val("");
        // 隐藏进度条
        loading.baosight.hidePageLoadingMsg();
    });
}

// 连接节点
function conn() {
    // 显示进度条
    loading.baosight.showPageLoadingMsg(false);
    // 发起请求
    $.post("conn", function (data) {
        // 展示操作结果
        $("#result").html(data)
        // 隐藏进度条
        loading.baosight.hidePageLoadingMsg();
    });
}

// 广播
function broadcast() {
    // 获取用户输入的内容
    var msg = $("#node").val();
    // 显示进度条
    loading.baosight.showPageLoadingMsg(false);
    // 发起请求
    $.post("broadcast", "msg=" + msg, function (data) {
        // 展示操作结果
        $("#result").html(data)
        // 清空输入框
        $("#node").val("");
        // 隐藏进度条
         loading.baosight.hidePageLoadingMsg();
    });
}

（7）启动服务器测试。

第一步：点击Edit Configuration，把Single instance only的勾取消掉。

第二步：启动两个sprintboot，一个监听7000端口号，一个监听8000端口号。

第三步：启动成功后，在浏览器分别打开localhost:7000和localhost:8000两个页面进行测试。

测试流程：

1）在localhost:7000页面中先注册节点8001，然后点击连接。

2）在localhost:8000页面中先注册节点7001，然后点击连接。

3）在任意一个页面中广播消息，比如在localhost:7000页面广播了一条消息，那么在8000的springboot控制台上可以看到该条广播消息。

结果如下图所示：

 

二、同步数据

（1）在页面上添加同步按钮。

    注册节点
    连接节点
    广播
    同步

（2）定义事件函数。

// 同步
function syncData() {
   // 显示进度条
   loading.baosight.showPageLoadingMsg(false);
   // 发起请求
   $.post("syncData", function (data) {
        // 展示操作结果
        $("#result").html(data)
        // 显示数据
        showList();
        // 隐藏进度条
        loading.baosight.hidePageLoadingMsg();
   });
}

（3）在BlockController中定义syncData方法，执行同步操作。

// 请求同步其他节点的区块链数据
@RequestMapping(value = "/syncData")
public String syncData() {
    for (MyClient client : clients) {
        client.send("兄弟，请把您的区块链数据给我一份");
    }
    return "同步成功";
}

（4）修改MyServer的onMessage函数，该函数把该节点的区块链数据广播给所有已注册的节点。

@Override
public void onMessage(WebSocket conn, String message) {
    System.out.println("WebSocket_" + port + "接收到消息...");
    try {
        if ("兄弟，请把您的区块链数据给我一份".equals(message)) {
            // 获取区块连数据
            NoteBook noteBooke = NoteBook.newInstance();
            List blocks = noteBooke.showList();
            // 把blocks转换成字符串
            ObjectMapper objectMapper = new ObjectMapper();
            String blockInfos = objectMapper.writeValueAsString(blocks);
            // 把数据封装到MessageBean
            MessageBean mb = new MessageBean(1, blockInfos);
            // 把MessageBean对象转换成字符串
            String msg = objectMapper.writeValueAsString(mb);
            // 广播消息
            broadcast(msg);
        }
    } catch (JsonProcessingException e) {
        e.printStackTrace();
    }
}

（5）修改MyClient的onMessage函数，该函数用于同步注册节点的区块链数据。

@Override
public void onMessage(String message) {
    System.out.println("客户端" + name + "收到服务端发送过来的消息...");
    try {
        ObjectMapper objectMapper = new ObjectMapper();
        // 把message转换成MessageBean对象
        MessageBean mb = objectMapper.readValue(message, MessageBean.class);
        // 判断消息类型
        if (mb.getType() == 1) {
            // 把消息转换成ArrayList对象
            JavaType javaType = objectMapper.getTypeFactory().constructParametricType(ArrayList.class, Block.class);
            ArrayList newList = objectMapper.readValue(mb.getMsg(), javaType);
            // 比较本地blocks和得到的blocks的长度
            // 如果本地blocks的长度比得到blocks的长度小，代表需要同步。
            NoteBook noteBook = NoteBook.newInstance();
            noteBook.compareData(newList);
        }
    } catch (IOException e) {
        e.printStackTrace();
    }
}

（6）定义MessageBean类，用于不同类型的数据。

public class MessageBean {
    private int type; // 1代表区块链数据 2代表交易数据
    private String msg; // 消息内容

    public MessageBean() {}

    public MessageBean(int type, String msg) {
        this.type = type;
        this.msg = msg;
    }

    ...
}

（7）修改NoteBook类，把NoteBook对象定义成单例模式。

private static NoteBook noteBook;

public static NoteBook newInstance() {
    if (noteBook == null) {
        synchronized (NoteBook.class) {
            if (noteBook == null) {
                noteBook = new NoteBook();
            }
        }
    }
    return noteBook;
}

（7）测试程序。

启动两个springboot，在浏览器打开两个页面。先执行注册和连接操作，然后在一个页面中添加区块， 在另外一个页面中执行同步操作。如果同步成功，可以在页面上看到刚才添加的区块数据。

 

三、广播交易数据

（1）修改BlockController的addBlock函数，添加区块的时候进行广播操作。

@RequestMapping(value = "/addBlock", method = RequestMethod.POST)
public String addBlock(Transaction tx) {
    try {
        if (tx.verify()) {
            // 把Transaction对象转换成字符串
            ObjectMapper objectMapper = new ObjectMapper();
            String txInfo = objectMapper.writeValueAsString(tx);
            // 把交易数据封装成MessageBean对象
            MessageBean mb = new MessageBean(2, txInfo);
            String msg = objectMapper.writeValueAsString(mb);
            broadcast(msg);
            // 执行添加操作
            noteBook.addBlock(txInfo);
            return "添加区块成功！";
        } else {
            throw new RuntimeException("交易数据校验失败！");
        }
    } catch (Exception e) {
        return "添加失败：" + e.getMessage();
    }
}

（2）修改MyClient的onMessage方法，同步交易数据。

@Override
public void onMessage(String message) {
    System.out.println("客户端" + name + "收到服务端发送过来的消息...");
    try {
        ObjectMapper objectMapper = new ObjectMapper();
        // 把message转换成MessageBean对象
        MessageBean mb = objectMapper.readValue(message, MessageBean.class);
        NoteBook noteBook = NoteBook.newInstance();
        // 判断消息类型
        if (mb.getType() == 1) {
            // 把消息转换成ArrayList对象
            JavaType javaType = objectMapper.getTypeFactory().constructParametricType(ArrayList.class, Block.class);
            ArrayList newList = objectMapper.readValue(mb.getMsg(), javaType);
            // 比较本地blocks和得到的blocks的长度
            // 如果本地blocks的长度比得到blocks的长度小，代表需要同步。
            noteBook.compareData(newList);
        } else if (mb.getType() == 2) {
            // 把msg数据转换成Transaction对象
            Transaction tx = objectMapper.readValue(mb.getMsg(), Transaction.class);
            // 交易交易
            if (tx.verify()) {
                noteBook.addBlock(mb.getMsg());
            }
        }
    } catch (IOException e) {
        e.printStackTrace();
    }
}

最后运行测试即可。同样也是启动两个springboot，在浏览器打开两个页面。先执行注册和连接操作。然后在一个页面中添加区块， 在另外一个页面刷新数据。这样就可以看到刚才添加的区块数据。