golang p2p網

繼續進入下一個初始化

南康ssl適用于網站、小程序/APP、API接口等需要進行數據傳輸應用場景，ssl證書未來市場廣闊！成為創新互聯建站的ssl證書銷售渠道，可以享受市場價格4-6折優惠！如果有意向歡迎電話聯系或者加微信：18980820575（備注：SSL證書合作）期待與您的合作！

n.netService, err = nebnet.NewNebService(n)

if err != nil {

logging.CLog().WithFields(logrus.Fields{

"err": err,

}).Fatal("Failed to setup net service.")

}

netservice有兩個成員

type NebServicestruct {

node? ? ? *Node

dispatcher *Dispatcher

}

跳出stup()函數

先進入start()函數看一看

if err := n.netService.Start(); err != nil {

logging.CLog().WithFields(logrus.Fields{

"err": err,

}).Fatal("Failed to start net service.")

}

進入netservice.start()

func (ns *NebService) Start() error {

logging.CLog().Info("Starting NebService...")

// start dispatcher.

ns.dispatcher.Start()

// start node.

if err := ns.node.Start(); err != nil {

ns.dispatcher.Stop()

logging.CLog().WithFields(logrus.Fields{

"err": err,

}).Error("Failed to start NebService.")

return err

}

logging.CLog().Info("Started NebService.")

return nil

}

可以看到第一個start()的函數是dispatcher.start()

進入dispatch.start()

func (dp *Dispatcher) Start() {

logging.CLog().Info("Starting NebService Dispatcher...")

go dp.loop()

}

然后就出現一個新的線程、goruntime

go dp.loop()

進入該線程，看它干了些什么

timerChan := time.NewTicker(time.Second).C

for {

select {

case -timerChan:

metricsDispatcherCached.Update(int64(len(dp.receivedMessageCh)))

case -dp.quitCh:

logging.CLog().Info("Stoped NebService Dispatcher.")

return

case msg := -dp.receivedMessageCh:

msgType := msg.MessageType()

v, _ := dp.subscribersMap.Load(msgType)

if v == nil {

continue

? }

m, _ := v.(*sync.Map)

m.Range(func(key, valueinterface{}) bool {

select {

case key.(*Subscriber).msgChan - msg:

default:

logging.VLog().WithFields(logrus.Fields{

"msgType": msgType,

}).Warn("timeout to dispatch message.")

}

return true

? })

}

}

一個有點長的循環

metricsDispatcherCached.Update(int64(len(dp.receivedMessageCh)))一秒鐘刷新一次緩沖區

case msg := -dp.receivedMessageCh:

msgType := msg.MessageType()如果能取出dp.receivedMessageCh

msgType := msg.MessageType()首先判斷取出的信息類型

v, _ := dp.subscribersMap.Load(msgType)

if v == nil {

continue

}

根據類型取出相應的map

如果取不出，那么使用continue結束這個case

m, _ := v.(*sync.Map)

斷言

m.Range(func(key, valueinterface{}) bool {

select {

case key.(*Subscriber).msgChan - msg:

default:

logging.VLog().WithFields(logrus.Fields{

"msgType": msgType,

}).Warn("timeout to dispa+tch message.")

}

return true

})

將msg推入其他管道里面去。其他goruntime會循環等待該

IPFS(四) 源碼解讀之-p2p

package p2p

import (

"context"

"errors"

"time"

net "gx/ipfs/QmPjvxTpVH8qJyQDnxnsxF9kv9jezKD1kozz1hs3fCGsNh/go-libp2p-net"

manet "gx/ipfs/QmV6FjemM1K8oXjrvuq3wuVWWoU2TLDPmNnKrxHzY3v6Ai/go-multiaddr-net"

ma "gx/ipfs/QmYmsdtJ3HsodkePE3eU3TsCaP2YvPZJ4LoXnNkDE5Tpt7/go-multiaddr"

pro "gx/ipfs/QmZNkThpqfVXs9GNbexPrfBbXSLNYeKrE7jwFM2oqHbyqN/go-libp2p-protocol"

pstore "gx/ipfs/QmZR2XWVVBCtbgBWnQhWk2xcQfaR3W8faQPriAiaaj7rsr/go-libp2p-peerstore"

p2phost "gx/ipfs/Qmb8T6YBBsjYsVGfrihQLfCJveczZnneSBqBKkYEBWDjge/go-libp2p-host"

peer "gx/ipfs/QmdVrMn1LhB4ybb8hMVaMLXnA8XRSewMnK6YqXKXoTcRvN/go-libp2p-peer"

)

//P2P結構保存當前正在運行的流/監聽器的信息

// P2P structure holds information on currently running streams/listeners

type P2P struct {

//監聽器

Listeners ListenerRegistry

//數據流

Streams StreamRegistry

//節點ID

identity peer.ID

//節點地址

peerHost p2phost.Host

//一個線程安全的對等節點存儲

peerstore pstore.Peerstore

}

//創建一個新的p2p結構

// NewP2P creates new P2P struct

//這個新的p2p結構不包含p2p結構中的監聽器和數據流

func NewP2P(identity peer.ID, peerHost p2phost.Host, peerstore pstore.Peerstore) *P2P {

return P2P{

identity: identity,

peerHost: peerHost,

peerstore: peerstore,

}

}

//新建一個數據流 工具方法 構建一個有節點id，內容和協議的流

func (p2p P2P) newStreamTo(ctx2 context.Context, p peer.ID, protocol string) (net.Stream, error) {

//30s 后會自動timeout

ctx, cancel := context.WithTimeout(ctx2, time.Second 30) //TODO: configurable?

defer cancel()

err := p2p.peerHost.Connect(ctx, pstore.PeerInfo{ID: p})

if err != nil {

return nil, err

}

return p2p.peerHost.NewStream(ctx2, p, pro.ID(protocol))

}

//對話為遠程監聽器創建新的P2P流

//創建一個新的p2p流實現對對話的監聽

// Dial creates new P2P stream to a remote listener

//Multiaddr是一種跨協議、跨平臺的表示格式的互聯網地址。它強調明確性和自我描述。

//對內接收

func (p2p P2P) Dial(ctx context.Context, addr ma.Multiaddr, peer peer.ID, proto string, bindAddr ma.Multiaddr) ( ListenerInfo, error) {

//獲取一些節點信息 network, host, nil

lnet, _, err := manet.DialArgs(bindAddr)

if err != nil {

return nil, err

}

//監聽信息

listenerInfo := ListenerInfo{

//節點身份

Identity: p2p.identity,

////應用程序協議標識符。

Protocol: proto,

}

//調用newStreamTo 通過ctx（內容） peer（節點id） proto（協議標識符） 參數獲取一個新的數據流

remote, err := p2p.newStreamTo(ctx, peer, proto)

if err != nil {

return nil, err

}

//network協議標識

switch lnet {

//network為"tcp", "tcp4", "tcp6"

case "tcp", "tcp4", "tcp6":

//從監聽器獲取新的信息 nla.Listener, nil

listener, err := manet.Listen(bindAddr)

if err != nil {

if err2 := remote.Reset(); err2 != nil {

return nil, err2

}

return nil, err

}

//將獲取的新信息保存到listenerInfo

listenerInfo.Address = listener.Multiaddr()

listenerInfo.Closer = listener

listenerInfo.Running = true

//開啟接受

go p2p.doAccept(listenerInfo, remote, listener)

default:

return nil, errors.New("unsupported protocol: " + lnet)

}

return listenerInfo, nil

}

//

func (p2p *P2P) doAccept(listenerInfo *ListenerInfo, remote net.Stream, listener manet.Listener) {

//關閉偵聽器并刪除流處理程序

defer listener.Close()

//Returns a Multiaddr friendly Conn

//一個有好的 Multiaddr 連接

local, err := listener.Accept()

if err != nil {

return

}

stream := StreamInfo{

//連接協議

Protocol: listenerInfo.Protocol,

//定位節點

LocalPeer: listenerInfo.Identity,

//定位節點地址

LocalAddr: listenerInfo.Address,

//遠程節點

RemotePeer: remote.Conn().RemotePeer(),

//遠程節點地址

RemoteAddr: remote.Conn().RemoteMultiaddr(),

//定位

Local: local,

//遠程

Remote: remote,

//注冊碼

Registry: p2p.Streams,

}

//注冊連接信息

p2p.Streams.Register(stream)

//開啟節點廣播

stream.startStreaming()

}

//偵聽器將流處理程序包裝到偵聽器中

// Listener wraps stream handler into a listener

type Listener interface {

Accept() (net.Stream, error)

Close() error

}

//P2PListener保存關于偵聽器的信息

// P2PListener holds information on a listener

type P2PListener struct {

peerHost p2phost.Host

conCh chan net.Stream

proto pro.ID

ctx context.Context

cancel func()

}

//等待偵聽器的連接

// Accept waits for a connection from the listener

func (il *P2PListener) Accept() (net.Stream, error) {

select {

case c := -il.conCh:

return c, nil

case -il.ctx.Done():

return nil, il.ctx.Err()

}

}

//關閉偵聽器并刪除流處理程序

// Close closes the listener and removes stream handler

func (il *P2PListener) Close() error {

il.cancel()

il.peerHost.RemoveStreamHandler(il.proto)

return nil

}

// Listen創建新的P2PListener

// Listen creates new P2PListener

func (p2p P2P) registerStreamHandler(ctx2 context.Context, protocol string) ( P2PListener, error) {

ctx, cancel := context.WithCancel(ctx2)

list := P2PListener{

peerHost: p2p.peerHost,

proto: pro.ID(protocol),

conCh: make(chan net.Stream),

ctx: ctx,

cancel: cancel,

}

p2p.peerHost.SetStreamHandler(list.proto, func(s net.Stream) {

select {

case list.conCh - s:

case -ctx.Done():

s.Reset()

}

})

return list, nil

}

// NewListener創建新的p2p偵聽器

// NewListener creates new p2p listener

//對外廣播

func (p2p P2P) NewListener(ctx context.Context, proto string, addr ma.Multiaddr) ( ListenerInfo, error) {

//調用registerStreamHandler 構造一個新的listener

listener, err := p2p.registerStreamHandler(ctx, proto)

if err != nil {

return nil, err

}

//構造新的listenerInfo

listenerInfo := ListenerInfo{

Identity: p2p.identity,

Protocol: proto,

Address: addr,

Closer: listener,

Running: true,

Registry: p2p.Listeners,

}

go p2p.acceptStreams(listenerInfo, listener)

//注冊連接信息

p2p.Listeners.Register(listenerInfo)

return listenerInfo, nil

}

//接受流

func (p2p *P2P) acceptStreams(listenerInfo *ListenerInfo, listener Listener) {

for listenerInfo.Running {

//一個有好的 遠程 連接

remote, err := listener.Accept()

if err != nil {

listener.Close()

break

}

}

//取消注冊表中的p2p偵聽器

p2p.Listeners.Deregister(listenerInfo.Protocol)

}

// CheckProtoExists檢查是否注冊了協議處理程序

// mux處理程序

// CheckProtoExists checks whether a protocol handler is registered to

// mux handler

func (p2p *P2P) CheckProtoExists(proto string) bool {

protos := p2p.peerHost.Mux().Protocols()

for _, p := range protos {

if p != proto {

continue

}

return true

}

return false

}

go語言能做什么？

很多朋友可能知道Go語言的優勢在哪，卻不知道Go語言適合用于哪些地方。

1、 Go語言作為服務器編程語言，很適合處理日志、數據打包、虛擬機處理、文件系統、分布式系統、數據庫代理等；網絡編程方面。Go語言廣泛應用于Web應用、API應用、下載應用等；除此之外，Go語言還可用于內存數據庫和云平臺領域，目前國外很多云平臺都是采用Go開發。

2、 其實Go語言主要用作服務器端開發。其定位是用來開發"大型軟件"的，適合于很多程序員一起開發大型軟件，并且開發周期長，支持云計算的網絡服務。Go語言能夠讓程序員快速開發，并且在軟件不斷的增長過程中，它能讓程序員更容易地進行維護和修改。它融合了傳統編譯型語言的高效性和腳本語言的易用性和富于表達性。

3、 Go語言成功案例。Nsq：Nsq是由Go語言開發的高性能、高可用消息隊列系統，性能非常高，每天能處理數十億條的消息；

4、 Docker：基于lxc的一個虛擬打包工具，能夠實現PAAS平臺的組建。

5、 Packer：用來生成不同平臺的鏡像文件，例如VM、vbox、AWS等，作者是vagrant的作者

6、 Skynet：分布式調度框架。

7、 Doozer：分布式同步工具，類似ZooKeeper。

8、 Heka：mazila開源的日志處理系統。

9、 Cbfs：couchbase開源的分布式文件系統。

10、 Tsuru：開源的PAAS平臺，和SAE實現的功能一模一樣。

11、 Groupcache：memcahe作者寫的用于Google下載系統的緩存系統。

12、 God：類似redis的緩存系統，但是支持分布式和擴展性。

13、 Gor：網絡流量抓包和重放工具。

以上的就是關于go語言能做什么的內容介紹了。

Go語言設計與實現（上）

基本設計思路：

類型轉換、類型斷言、動態派發。iface，eface。

反射對象具有的方法：

編譯優化：

內部實現：

實現 Context 接口有以下幾個類型（空實現就忽略了）：

互斥鎖的控制邏輯：

設計思路：

（以上為寫被讀阻塞，下面是讀被寫阻塞）

總結，讀寫鎖的設計還是非常巧妙的：

設計思路：

WaitGroup 有三個暴露的函數:

部件：

設計思路：

結構：

Once 只暴露了一個方法：

實現：

三個關鍵點：

細節：

讓多協程任務的開始執行時間可控（按順序或歸一）。（Context 是控制結束時間）

設計思路： 通過一個鎖和內置的 notifyList 隊列實現，Wait() 會生成票據，并將等待協程信息加入鏈表中，等待控制協程中發送信號通知一個（Signal()）或所有（Boardcast()）等待者（內部實現是通過票據通知的）來控制協程解除阻塞。

暴露四個函數：

實現細節：

部件：

包： golang.org/x/sync/errgroup

作用：開啟 func() error 函數簽名的協程，在同 Group 下協程并發執行過程并收集首次 err 錯誤。通過 Context 的傳入，還可以控制在首次 err 出現時就終止組內各協程。

設計思路：

結構：

暴露的方法：

實現細節：

注意問題：

包： "golang.org/x/sync/semaphore"

作用：排隊借資源（如錢，有借有還）的一種場景。此包相當于對底層信號量的一種暴露。

設計思路：有一定數量的資源 Weight，每一個 waiter 攜帶一個 channel 和要借的數量 n。通過隊列排隊執行借貸。

結構：

暴露方法：

細節：

部件：

細節：

包： "golang.org/x/sync/singleflight"

作用：防擊穿。瞬時的相同請求只調用一次，response 被所有相同請求共享。

設計思路：按請求的 key 分組（一個 *call 是一個組，用 map 映射存儲組），每個組只進行一次訪問，組內每個協程會獲得對應結果的一個拷貝。

結構：

邏輯：

細節：

部件：

如有錯誤，請批評指正。

新聞標題：p2pgo語言實現 p2p編程源碼
分享網址：http://vcdvsql.cn/article24/dopehje.html

成都網站建設公司_創新互聯，為您提供網站制作、品牌網站建設、網站改版、ChatGPT、外貿網站建設、品牌網站設計

廣告

聲明：本網站發布的內容（圖片、視頻和文字）以用戶投稿、用戶轉載內容為主，如果涉及侵權請盡快告知，我們將會在第一時間刪除。文章觀點不代表本網站立場，如需處理請聯系客服。電話：028-86922220；郵箱：631063699@qq.com。內容未經允許不得轉載，或轉載時需注明來源： 創新互聯