使用Go语言编写区块链P2P网络(译)

在之前的文章中，我们已经知道了怎么编写PoW也知道了IPFS怎么工作, 但是有一个致命的缺点，我们的服务都是中心化的，这篇文章会教你怎么实现一个简单的完全去中心化的P2P网络。

背景知识

什么是P2P网络

在真正的P2P架构中，不需要中心化的服务来维护区块链的状态。例如，当你给朋友发送比特币时，比特币区块链的“状态”应该更新，这样你朋友的余额就会增加，你的余额就会减少。

在这个网络中，不存在一个权力高度中心化的机构来维护状态（银行就是这样的中心化机构)。对于比特币网络来说，每个节点都会维护一份完整的区块链状态，当交易发生时，每个节点的区块链状态都会得到更新。这样，只要网络中51%的节点对区块链的状态达成一致，那么区块链网络就是安全可靠的，具体可以阅读这篇一致性协议文章。

本文将继续之前的工作,200行Go代码实现区块链, 并加入P2P网络架构。在继续之前，强烈建议你先阅读该篇文章，它会帮助你理解接下来的代码。

开始实现

编写P2P网络可不是开开玩笑就能简单视线的，有很多边边角角的情况都要覆盖到，而且需要你拥有很多工程学的知识，这样的P2P网络才是可扩展、高可靠的。有句谚语说得好：站在巨人肩膀上做事，那么我们先看看巨人们提供了哪些工具吧。

喔，看看，我们发现了什么！一个用Go语言实现的P2P库go-libp2p！如果你对新技术足够敏锐，就会发现这个库的作者和IPFS的作者是同一个团队。如果你还没看过我们的IPFS教程，可以看看这里, 你可以选择跳过IPFS教程，因为对于本文这不是必须的。

警告

目前来说， go-libp2p主要有两个缺点:

1. 安装设置比较痛苦，它使用gx作为包管理工具，怎么说呢，不咋好用，但是凑活用吧
   
2. 目前项目还没有成熟，正在紧密锣鼓的开发中，当使用这个库时，可能会遇到一些数据竞争(data race)
   

对于第一点，不必担心，有我们呢。第二点是比较大的问题，但是不会影响我们的代码。假如你在使用过程中发现了数据竞争问题，记得给项目提一个issue，帮助它更好的成长！

总之，目前开源世界中，现代化的P2P库是非常非常少的，因为我们要多给 go-libp2p一些耐心和包容，而且就目前来说，它已经能很好的满足我们的目标了。

安装设置

最好的环境设置方式是直接clone libp2p库，然后在这个库的代码中直接开发。你也可以在自己的库中，调用这个库开发，但是这样就需要用到 gx了。这里我们使用简单的方式，假设你已经安装了Go:

• go get-d github.com/libp2p/go-libp2p/…
  
• 进入 go-libp2p文件夹
  
• make
  
• make deps
  

这里会通过gx包管理工具下载所有需要的包和依赖，再次申明，我们不喜欢gx，因为它打破了Go语言的很多惯例，但是为了这个很棒的库，认怂吧。

这里，我们在 examples子目录下进行开发，因此在 go-libp2p的examples下创建一个你自己的目录

• mkdir ./examples/p2p
  

然后进入到p2p文件夹下，创建 main.go文件，后面所有的代码都会在该文件中。

你的目录结构是这样的： 好了，勇士们，拔出你们的剑，哦不，拔出你们的 main.go，开始我们的征途吧！

导入相关库

这里申明我们需要用的库，大部分库是来自于 go-libp2p本身的，在教程中，你会学到怎么去使用它们。

1. package main
   
2. 
   
3. import(
   
4.    "bufio"
   
5.    "context"
   
6.    "crypto/rand"
   
7.    "crypto/sha256"
   
8.    "encoding/hex"
   
9.    "encoding/json"
   
10.    "flag"
    
11.    "fmt"
    
12.    "io"
    
13.    "log"
    
14.    mrand "math/rand"
    
15.    "os"
    
16.    "strconv"
    
17.    "strings"
    
18.    "sync"
    
19.    "time"
    
20. 
    
21.    "github.com/davecgh/go-spew/spew"
    
22.    golog "github.com/ipfs/go-log"
    
23.    libp2p "github.com/libp2p/go-libp2p"
    
24.    crypto "github.com/libp2p/go-libp2p-crypto"
    
25.    host "github.com/libp2p/go-libp2p-host"
    
26.    net "github.com/libp2p/go-libp2p-net"
    
27.    peer "github.com/libp2p/go-libp2p-peer"
    
28.    pstore "github.com/libp2p/go-libp2p-peerstore"
    
29.    ma "github.com/multiformats/go-multiaddr"
    
30.    gologging "github.com/whyrusleeping/go-logging"
    
31. )
    

spew包可以很方便、优美的打印出我们的区块链，因此记得安装它：

• go getgithub.com/davecgh/go-spew/spew
  

区块链结构

记住，请先阅读200行Go代码实现区块链， 这样，下面的部分就会简单很多。

先来申明全局变量：

1. // Block represents each 'item' in the blockchain
   
2. type Blockstruct{
   
3.    Index     int
   
4.    Timestampstring
   
5.    BPM       int
   
6.    Hash      string
   
7.    PrevHash  string
   
8. }
   
9. 
   
10. // Blockchain is a series of validated Blocks
    
11. varBlockchain[]Block
    
12. 
    
13. var mutex =&sync.Mutex{}
    

• 我们是一家健康看护公司，因此Block中存着的是用户的脉搏速率BPM
  
• Blockchain是我们的"状态"，或者严格的说：最新的Blockchain，它其实就是Block的切片（slice)
  
• mutex是为了防止资源竞争出现
  

下面是Blockchain相关的特定函数：

1. // make sure block is valid by checking index, and comparing the hash of the previous block
   
2. func isBlockValid(newBlock, oldBlock Block)bool{
   
3.    if oldBlock.Index+1!= newBlock.Index{
   
4.        returnfalse
   
5.    }
   
6. 
   
7.    if oldBlock.Hash!= newBlock.PrevHash{
   
8.        returnfalse
   
9.    }
   
10. 
    
11.    if calculateHash(newBlock)!= newBlock.Hash{
    
12.        returnfalse
    
13.    }
    
14. 
    
15.    returntrue
    
16. }
    
17. 
    
18. // SHA256 hashing
    
19. func calculateHash(block Block)string{
    
20.    record := strconv.Itoa(block.Index)+ block.Timestamp+ strconv.Itoa(block.BPM)+ block.PrevHash
    
21.    h := sha256.New()
    
22.    h.Write([]byte(record))
    
23.    hashed := h.Sum(nil)
    
24.    return hex.EncodeToString(hashed)
    
25. }
    
26. 
    
27. // create a new block using previous block's hash
    
28. func generateBlock(oldBlock Block, BPM int)Block{
    
29. 
    
30.    var newBlock Block
    
31. 
    
32.    t := time.Now()
    
33. 
    
34.    newBlock.Index= oldBlock.Index+1
    
35.    newBlock.Timestamp= t.String()
    
36.    newBlock.BPM = BPM
    
37.    newBlock.PrevHash= oldBlock.Hash
    
38.    newBlock.Hash= calculateHash(newBlock)
    
39. 
    
40.    return newBlock
    
41. }
    

• isBlockValid检查Block的hash是否合法
  
• calculateHash使用 sha256来对原始数据做hash
  
• generateBlock创建一个新的Block区块，然后添加到区块链Blockchain上，同时会包含所需的事务
  

P2P结构

下面我们快接近核心部分了，首先我们要写出创建主机的逻辑。当一个节点运行我们的程序时，它可以作为一个主机，被其它节点连接。下面一起看看代码:-)

1. // makeBasicHost creates a LibP2P host with a random peer ID listening on the
   
2. // given multiaddress. It will use secio if secio is true.
   
3. func makeBasicHost(listenPort int, secio bool, randseed int64)(host.Host, error){
   
4. 
   
5.    // If the seed is zero, use real cryptographic randomness. Otherwise, use a
   
6.    // deterministic randomness source to make generated keys stay the same
   
7.    // across multiple runs
   
8.    var r io.Reader
   
9.    if randseed ==0{
   
10.        r = rand.Reader
    
11.    }else{
    
12.        r = mrand.New(mrand.NewSource(randseed))
    
13.    }
    
14. 
    
15.    // Generate a key pair for this host. We will use it
    
16.    // to obtain a valid host ID.
    
17.    priv, _, err := crypto.GenerateKeyPairWithReader(crypto.RSA,2048, r)
    
18.    if err !=nil{
    
19.        returnnil, err
    
20.    }
    
21. 
    
22.    opts :=[]libp2p.Option{
    
23.        libp2p.ListenAddrStrings(fmt.Sprintf("/ip4/127.0.0.1/tcp/%d", listenPort)),
    
24.        libp2p.Identity(priv),
    
25.    }
    
26. 
    
27.    if!secio {
    
28.        opts = append(opts, libp2p.NoEncryption())
    
29.    }
    
30. 
    
31.    basicHost, err := libp2p.New(context.Background(), opts...)
    
32.    if err !=nil{
    
33.        returnnil, err
    
34.    }
    
35. 
    
36.    // Build host multiaddress
    
37.    hostAddr, _ := ma.NewMultiaddr(fmt.Sprintf("/ipfs/%s", basicHost.ID().Pretty()))
    
38. 
    
39.    // Now we can build a full multiaddress to reach this host
    
40.    // by encapsulating both addresses:
    
41.    addr := basicHost.Addrs()[0]
    
42.    fullAddr := addr.Encapsulate(hostAddr)
    
43.    log.Printf("I am %s\n", fullAddr)
    
44.    if secio {
    
45.        log.Printf("Now run \"go run main.go -l %d -d %s -secio\" on a different terminal\n", listenPort+1, fullAddr)
    
46.    }else{
    
47.        log.Printf("Now run \"go run main.go -l %d -d %s\" on a different terminal\n", listenPort+1, fullAddr)
    
48.    }
    
49. 
    
50.    return basicHost,nil
    
51. }
    

makeBasicHost函数有3个参数，同时返回一个host结构体

• listenPort是主机监听的端口，其它节点会连接该端口
  
• secio表明是否开启数据流的安全选项，最好开启，因此它代表了"安全输入／输出"
  
• randSeed是一个可选的命令行标识，可以允许我们提供一个随机数种子来为我们的主机生成随机的地址。这里我们不会使用
  

函数的第一个 if语句针对随机种子生成随机key，接着我们生成公钥和私钥，这样能保证主机是安全的。 opts部分开始构建网络地址部分，这样其它节点就可以连接进来。

!secio部分可以绕过加密，但是我们准备使用加密，因此这段代码不会被触发。

接着，创建了主机地址，这样其他节点就可以连接进来。 log.Printf可以用来在控制台打印出其它节点的连接信息。最后我们返回生成的主机地址给调用方函数。

流处理

之前的主机需要能处理进入的数据流。当另外一个节点连接到主机时，它会想要提出一个新的区块链，来覆盖主机上的区块链，因此我们需要逻辑来判定是否要接受新的区块链。

同时，当我们往本地的区块链添加区块后，也要把相关信息广播给其它节点，这里也需要实现相关逻辑。

先来创建流处理的基本框架吧：

1. func handleStream(s net.Stream){
   
2. 
   
3.    log.Println("Got a new stream!")
   
4. 
   
5.    // Create a buffer stream for non blocking read and write.
   
6.    rw := bufio.NewReadWriter(bufio.NewReader(s), bufio.NewWriter(s))
   
7. 
   
8.    go readData(rw)
   
9.    go writeData(rw)
   
10. 
    
11.    // stream 's' will stay open until you close it (or the other side closes it).
    
12. }
    

这里创建一个新的 ReadWriter，为了能支持数据读取和写入，同时我们启动了一个单独的Go协程来处理相关读写逻辑。

读取数据

首先创建 readData函数：

1. func readData(rw *bufio.ReadWriter){
   
2. 
   
3.    for{
   
4.        str, err := rw.ReadString('\n')
   
5.        if err !=nil{
   
6.            log.Fatal(err)
   
7.        }
   
8. 
   
9.        if str ==""{
   
10.            return
    
11.        }
    
12.        if str !="\n"{
    
13. 
    
14.            chain := make([]Block,0)
    
15.            if err := json.Unmarshal([]byte(str),&chain); err !=nil{
    
16.                log.Fatal(err)
    
17.            }
    
18. 
    
19.            mutex.Lock()
    
20.            if len(chain)> len(Blockchain){
    
21.                Blockchain= chain
    
22.                bytes, err := json.MarshalIndent(Blockchain,"","  ")
    
23.                if err !=nil{
    
24. 
    
25.                    log.Fatal(err)
    
26.                }
    
27.                // Green console color:     \x1b[32m
    
28.                // Reset console color:     \x1b[0m
    
29.                fmt.Printf("\x1b[32m%s\x1b[0m> ",string(bytes))
    
30.            }
    
31.            mutex.Unlock()
    
32.        }
    
33.    }
    
34. }
    

该函数是一个无限循环，因为它需要永不停歇的去读取外面进来的数据。首先，我们使用 ReadString解析从其它节点发送过来的新的区块链（JSON字符串)。

然后检查进来的区块链的长度是否比我们本地的要长，如果进来的链更长，那么我们就接受新的链为最新的网络状态（最新的区块链)。

同时，把最新的区块链在控制台使用一种特殊的颜色打印出来，这样我们就知道有新链接受了。

如果在我们主机的本地添加了新的区块到区块链上，那就需要把本地最新的区块链广播给其它相连的节点知道，这样这些节点机会接受并更新到我们的区块链版本。这里使用 writeData函数:

1. func writeData(rw *bufio.ReadWriter){
   
2. 
   
3.    go func(){
   
4.        for{
   
5.            time.Sleep(5* time.Second)
   
6.            mutex.Lock()
   
7.            bytes, err := json.Marshal(Blockchain)
   
8.            if err !=nil{
   
9.                log.Println(err)
   
10.            }
    
11.            mutex.Unlock()
    
12. 
    
13.            mutex.Lock()
    
14.            rw.WriteString(fmt.Sprintf("%s\n",string(bytes)))
    
15.            rw.Flush()
    
16.            mutex.Unlock()
    
17. 
    
18.        }
    
19.    }()
    
20. 
    
21.    stdReader := bufio.NewReader(os.Stdin)
    
22. 
    
23.    for{
    
24.        fmt.Print("> ")
    
25.        sendData, err := stdReader.ReadString('\n')
    
26.        if err !=nil{
    
27.            log.Fatal(err)
    
28.        }
    
29. 
    
30.        sendData = strings.Replace(sendData,"\n","",-1)
    
31.        bpm, err := strconv.Atoi(sendData)
    
32.        if err !=nil{
    
33.            log.Fatal(err)
    
34.        }
    
35.        newBlock := generateBlock(Blockchain[len(Blockchain)-1], bpm)
    
36. 
    
37.        if isBlockValid(newBlock,Blockchain[len(Blockchain)-1]){
    
38.            mutex.Lock()
    
39.            Blockchain= append(Blockchain, newBlock)
    
40.            mutex.Unlock()
    
41.        }
    
42. 
    
43.        bytes, err := json.Marshal(Blockchain)
    
44.        if err !=nil{
    
45.            log.Println(err)
    
46.        }
    
47. 
    
48.        spew.Dump(Blockchain)
    
49. 
    
50.        mutex.Lock()
    
51.        rw.WriteString(fmt.Sprintf("%s\n",string(bytes)))
    
52.        rw.Flush()
    
53.        mutex.Unlock()
    
54.    }
    
55. 
    
56. }
    

首先是一个单独协程中的函数，每5秒钟会将我们的最新的区块链状态广播给其它相连的节点。它们收到后，如果发现我们的区块链比它们的要短，就会直接把我们发送的区块链信息丢弃，继续使用它们的区块链，反之则使用我们的区块链。总之，无论哪种方法，所有的节点都会定期的同步本地的区块链到最新状态。

这里我们需要一个方法来创建一个新的Block区块，包含之前提到过的脉搏速率（BPM)。为了简化实现，我们不会真的去通过物联网设备读取脉搏，而是直接在终端控制台上输入一个脉搏速率数字。

首先要验证输入的BPM是一个整数类型，然后使用之前的 generateBlock来生成区块，接着使用 spew.Dump输入到终端控制台，最后我们使用 rw.WriteString把最新的区块链广播给相连的其它节点。

牛逼了我的哥，现在我们完成了区块链相关的函数以及大多数P2P相关的函数。在前面，我们创建了流处理，因此可以读取和写入最新的区块链状态；创建了状态同步函数，这样节点之间可以互相同步最新状态。

剩下的就是实现我们的 main函数了：

1. func main(){
   
2.    t := time.Now()
   
3.    genesisBlock :=Block{}
   
4.    genesisBlock =Block{0, t.String(),0, calculateHash(genesisBlock),""}
   
5. 
   
6.    Blockchain= append(Blockchain, genesisBlock)
   
7. 
   
8.    // LibP2P code uses golog to log messages. They log with different
   
9.    // string IDs (i.e. "swarm"). We can control the verbosity level for
   
10.    // all loggers with:
    
11.    golog.SetAllLoggers(gologging.INFO)// Change to DEBUG for extra info
    
12. 
    
13.    // Parse options from the command line
    
14.    listenF := flag.Int("l",0,"wait for incoming connections")
    
15.    target := flag.String("d","","target peer to dial")
    
16.    secio := flag.Bool("secio",false,"enable secio")
    
17.    seed := flag.Int64("seed",0,"set random seed for id generation")
    
18.    flag.Parse()
    
19. 
    
20.    if*listenF ==0{
    
21.        log.Fatal("Please provide a port to bind on with -l")
    
22.    }
    
23. 
    
24.    // Make a host that listens on the given multiaddress
    
25.    ha, err := makeBasicHost(*listenF,*secio,*seed)
    
26.    if err !=nil{
    
27.        log.Fatal(err)
    
28.    }
    
29. 
    
30.    if*target ==""{
    
31.        log.Println("listening for connections")
    
32.        // Set a stream handler on host A. /p2p/1.0.0 is
    
33.        // a user-defined protocol name.
    
34.        ha.SetStreamHandler("/p2p/1.0.0", handleStream)
    
35. 
    
36.        select{}// hang forever
    
37.        /**** This is where the listener code ends ****/
    
38.    }else{
    
39.        ha.SetStreamHandler("/p2p/1.0.0", handleStream)
    
40. 
    
41.        // The following code extracts target's peer ID from the
    
42.        // given multiaddress
    
43.        ipfsaddr, err := ma.NewMultiaddr(*target)
    
44.        if err !=nil{
    
45.            log.Fatalln(err)
    
46.        }
    
47. 
    
48.        pid, err := ipfsaddr.ValueForProtocol(ma.P_IPFS)
    
49.        if err !=nil{
    
50.            log.Fatalln(err)
    
51.        }
    
52. 
    
53.        peerid, err := peer.IDB58Decode(pid)
    
54.        if err !=nil{
    
55.            log.Fatalln(err)
    
56.        }
    
57. 
    
58.        // Decapsulate the /ipfs/<peerID> part from the target
    
59.        // /ip4/<a.b.c.d>/ipfs/<peer> becomes /ip4/<a.b.c.d>
    
60.        targetPeerAddr, _ := ma.NewMultiaddr(
    
61.            fmt.Sprintf("/ipfs/%s", peer.IDB58Encode(peerid)))
    
62.        targetAddr := ipfsaddr.Decapsulate(targetPeerAddr)
    
63. 
    
64.        // We have a peer ID and a targetAddr so we add it to the peerstore
    
65.        // so LibP2P knows how to contact it
    
66.        ha.Peerstore().AddAddr(peerid, targetAddr, pstore.PermanentAddrTTL)
    
67. 
    
68.        log.Println("opening stream")
    
69.        // make a new stream from host B to host A
    
70.        // it should be handled on host A by the handler we set above because
    
71.        // we use the same /p2p/1.0.0 protocol
    
72.        s, err := ha.NewStream(context.Background(), peerid,"/p2p/1.0.0")
    
73.        if err !=nil{
    
74.            log.Fatalln(err)
    
75.        }
    
76.        // Create a buffered stream so that read and writes are non blocking.
    
77.        rw := bufio.NewReadWriter(bufio.NewReader(s), bufio.NewWriter(s))
    
78. 
    
79.        // Create a thread to read and write data.
    
80.        go writeData(rw)
    
81.        go readData(rw)
    
82. 
    
83.        select{}// hang forever
    
84. 
    
85.    }
    
86. }
    

首先是创建一个创世区块（如果你读了200行Go代码实现你的区块链,这里就不会陌生)。

其次我们使用 go-libp2p的 SetAllLoggers日志函数来记录日志。

接着，设置了所有的命令行标识：

• secio之前有提到，是用来加密数据流的。在我们的程序中，一定要打开该标识
  
• target指明当前节点要连接到的主机地址
  
• listenF是当前节点的监听主机地址，这样其它节点就可以连接进来，记住，每个节点都有两个身份：主机和客户端， 毕竟P2P不是白叫的
  
• seed是随机数种子，用来创建主机地址时使用
  

然后，使用 makeBasicHost函数来创建一个新的主机地址，如果我们只想做主机不想做客户端（连接其它的主机)，就使用 if*target==“”。

接下来的几行，会从 target解析出我们要连接到的主机地址。然后把 peerID和主机目标地址 targetAddr添加到"store"中，这样就可以持续跟踪我们跟其它主机的连接信息，这里使用的是 ha.Peerstore().AddAddr函数。

接着我们使用 ha.NewStream连接到想要连接的节点上，同时为了能接收和发送最新的区块链信息，创建了 ReadWriter，同时使用一个Go协程来进行 readData和 writeData。

哇哦

终于完成了，写文章远比写代码累！我知道之前的内容有点难，但是相比P2P的复杂性来说，你能通过一个库来完成P2P网络，已经很牛逼了，所以继续加油！

完整代码

mycoralhealth/blockchain-tutorial

运行结果

现在让我们来试验一下，首先打开3个独立的终端窗口做为独立节点。

开始之前，请再次进入 go-libp2p的根目录运行一下 make deps，确保所有依赖都正常安装。

回到你的工作目录 examples/p2p，打开第一个终端窗口，输入 go run main.go-l10000-secio 接着，该我们输入BPM数据了，在第一个终端窗口中输入"70"，等几秒中，观察各个窗口的打印输出。 来看看发生了什么：终端1向本地的区块链添加了一个新的区块Block终端1向终端2广播该信息终端2将新的区块链跟本地的对比，发现终端1的更长，因此使用新的区块链替代了本地的区块链，然后将新的区块链广播给终端3同上，终端3也进行更新所有的3个终端节点都把区块链更新到了最新版本，同时没有使用任何外部的中心化服务，这就是P2P网络的力量！我们再往终端2的区块链中添加一个区块试试看，在终端2中输入"80" 结果忠诚的记录了我们的正确性，再一次欢呼吧！下一步先享受一下自己的工作，你刚用了区区几百行代码就实现了一个全功能的P2P网络！这不是开玩笑，P2P编程时非常复杂的，为什么之前没有相关的教程，就是因为太难了。但是，这里也有几个可以改进的地方，你可以挑战一下自己：之前提到过, go-libp2p是存在数据竞争的Bug的，因此如果你要在生产环境使用，需要格外小心。一旦发现Bug，请反馈给作者团队知道尝试将本文的P2P网络跟之前的共识协议结合，例如之前的文章PoW 和PoS (PoS是中文译文)添加持久化存储。截止目前，为了简化实现，我们没有实现持久化存储，因此节点关闭，数据就丢失了本文的代码没有在大量节点的环境下测试过，试着写一个脚本运行大量节点，看看性能会怎么变化。如果发现Bug记得给我们提交学习一下节点发现技术。新节点是怎么发现已经存在的节点的？这篇文章是一个很好的起点