golang p2p網(wǎng)

繼續(xù)進(jìn)入下一個(gè)初始化

網(wǎng)站建設(shè)哪家好，找成都創(chuàng)新互聯(lián)！專注于網(wǎng)頁設(shè)計(jì)、網(wǎng)站建設(shè)、微信開發(fā)、微信小程序、集團(tuán)企業(yè)網(wǎng)站建設(shè)等服務(wù)項(xiàng)目。為回饋新老客戶創(chuàng)新互聯(lián)還提供了桃山免費(fèi)建站歡迎大家使用！

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

if err != nil {

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

"err": err,

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

}

netservice有兩個(gè)成員

type NebServicestruct {

node? ? ? *Node

dispatcher *Dispatcher

}

跳出stup()函數(shù)

先進(jìn)入start()函數(shù)看一看

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

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

"err": err,

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

}

進(jìn)入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

}

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

進(jìn)入dispatch.start()

func (dp *Dispatcher) Start() {

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

go dp.loop()

}

然后就出現(xiàn)一個(gè)新的線程、goruntime

go dp.loop()

進(jìn)入該線程，看它干了些什么

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

? })

}

}

一個(gè)有點(diǎn)長的循環(huán)

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

case msg := -dp.receivedMessageCh:

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

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

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

if v == nil {

continue

}

根據(jù)類型取出相應(yīng)的map

如果取不出，那么使用continue結(jié)束這個(gè)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會循環(huán)等待該

p2p的具體使用方法

p2p終結(jié)者設(shè)置

P2P終結(jié)者如果你設(shè)置好了是完全可以可以控制其他電腦，下面圖片你看看

主要在控制規(guī)則上你先建一個(gè)時(shí)間規(guī)則選擇所以時(shí)間，然后你新建個(gè)控制規(guī)則時(shí)間計(jì)劃選擇剛才你建時(shí)間規(guī)則，然后下一步：上行、下行全設(shè)置為10，p2p全選，通訊工具你可以只給他保留QQ，其余全選，http和ftp的后zuei設(shè)置文件比較大的后綴視頻類如.avi .rmvb .rm 軟件如.rar .zip .exe 音樂如.mp3 .mp4 .wma ..................，下一步，你可以限制也可以不限制，接著端口你新建吧端口范圍設(shè)置成0000-9999其余可以不用填寫完成就可以.然后你把規(guī)則指派到你要控制的IP上，這樣的效果可以達(dá)100%

最新的版本是4.7

到這個(gè)網(wǎng)站下載

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結(jié)構(gòu)保存當(dāng)前正在運(yùn)行的流/監(jiān)聽器的信息

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

type P2P struct {

//監(jiān)聽器

Listeners ListenerRegistry

//數(shù)據(jù)流

Streams StreamRegistry

//節(jié)點(diǎn)ID

identity peer.ID

//節(jié)點(diǎn)地址

peerHost p2phost.Host

//一個(gè)線程安全的對等節(jié)點(diǎn)存儲

peerstore pstore.Peerstore

}

//創(chuàng)建一個(gè)新的p2p結(jié)構(gòu)

// NewP2P creates new P2P struct

//這個(gè)新的p2p結(jié)構(gòu)不包含p2p結(jié)構(gòu)中的監(jiān)聽器和數(shù)據(jù)流

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

return P2P{

identity: identity,

peerHost: peerHost,

peerstore: peerstore,

}

}

//新建一個(gè)數(shù)據(jù)流 工具方法 構(gòu)建一個(gè)有節(jié)點(diǎn)id，內(nèi)容和協(xié)議的流

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))

}

//對話為遠(yuǎn)程監(jiān)聽器創(chuàng)建新的P2P流

//創(chuàng)建一個(gè)新的p2p流實(shí)現(xiàn)對對話的監(jiān)聽

// Dial creates new P2P stream to a remote listener

//Multiaddr是一種跨協(xié)議、跨平臺的表示格式的互聯(lián)網(wǎng)地址。它強(qiáng)調(diào)明確性和自我描述。

//對內(nèi)接收

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

//獲取一些節(jié)點(diǎn)信息 network, host, nil

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

if err != nil {

return nil, err

}

//監(jiān)聽信息

listenerInfo := ListenerInfo{

//節(jié)點(diǎn)身份

Identity: p2p.identity,

////應(yīng)用程序協(xié)議標(biāo)識符。

Protocol: proto,

}

//調(diào)用newStreamTo 通過ctx（內(nèi)容） peer（節(jié)點(diǎn)id） proto（協(xié)議標(biāo)識符） 參數(shù)獲取一個(gè)新的數(shù)據(jù)流

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

if err != nil {

return nil, err

}

//network協(xié)議標(biāo)識

switch lnet {

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

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

//從監(jiān)聽器獲取新的信息 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) {

//關(guān)閉偵聽器并刪除流處理程序

defer listener.Close()

//Returns a Multiaddr friendly Conn

//一個(gè)有好的 Multiaddr 連接

local, err := listener.Accept()

if err != nil {

return

}

stream := StreamInfo{

//連接協(xié)議

Protocol: listenerInfo.Protocol,

//定位節(jié)點(diǎn)

LocalPeer: listenerInfo.Identity,

//定位節(jié)點(diǎn)地址

LocalAddr: listenerInfo.Address,

//遠(yuǎn)程節(jié)點(diǎn)

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

//遠(yuǎn)程節(jié)點(diǎn)地址

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

//定位

Local: local,

//遠(yuǎn)程

Remote: remote,

//注冊碼

Registry: p2p.Streams,

}

//注冊連接信息

p2p.Streams.Register(stream)

//開啟節(jié)點(diǎn)廣播

stream.startStreaming()

}

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

// Listener wraps stream handler into a listener

type Listener interface {

Accept() (net.Stream, error)

Close() error

}

//P2PListener保存關(guān)于偵聽器的信息

// 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()

}

}

//關(guān)閉偵聽器并刪除流處理程序

// Close closes the listener and removes stream handler

func (il *P2PListener) Close() error {

il.cancel()

il.peerHost.RemoveStreamHandler(il.proto)

return nil

}

// Listen創(chuàng)建新的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創(chuàng)建新的p2p偵聽器

// NewListener creates new p2p listener

//對外廣播

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

//調(diào)用registerStreamHandler 構(gòu)造一個(gè)新的listener

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

if err != nil {

return nil, err

}

//構(gòu)造新的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 {

//一個(gè)有好的 遠(yuǎn)程 連接

remote, err := listener.Accept()

if err != nil {

listener.Close()

break

}

}

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

p2p.Listeners.Deregister(listenerInfo.Protocol)

}

// CheckProtoExists檢查是否注冊了協(xié)議處理程序

// 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

}