• 资料库的webrtc文件传输

    一、一个看似简单的事情往往不简单

    一个简单的事情往往会倾注你的心血，也许你看到很简单往往其实没那么简单；其实想想今年业余时间的大把代码，真正能成品的好像并不多。

    马上年底了，写下这篇文章。每一行程序就像写小说一样，不管好不好；代码都倾注了我的心血。

    真正把一个东西做好；做到好用，用的人多，具有很好的利他性，并不是一个容易的事情，需要不断的量变。

    webrtc演示网址:

    文件下载测试

    二、为什么放弃百度网盘

    把数据存储到百度网盘上大该有半年时间，后来发现了一些致命的问题

    1、发现分享外链到了10几万条之后就不能分享了

    2、分享的百度外链经常会被屏蔽掉

    3、分享速度不能过快，1小时只能分享100多条外链，如果分享的过快整个网盘的分享功能都会

    4、百度账号需要实名认证，很容易被人认出来，作为一个技术人员我认为应该藏在幕后，尽量不要让别人认出来，这回避免很多不必要的问题

    三、为什么不用云存储

    4个T的存储，每月就要400元，一年就是5600元，而且不算流量费用一年就是5600，这个费用是不高的，最关键的是使用了云存储很

    1、可能被黑客共计，如果别人使用curl去恶意刷流量现有的安全策略很可能被人把费用都刷去

    2、每年存储费用是递增的

    3、需要实名认证，一旦绑定域名，绑定着手机号，很可能被人找到

    四、我的WebRtc架构设计

    拜托传统的中继服务器，使用p2p打洞，调研了一些方案，本来打算使用 coturn + janus 的架构设计，但是发现janus c 写的，没有文件传输的插件，如果自己再用c去做一个插件真的是很累，于是就打算用golang 实现一个文件传输的网关。

    大体步骤

    1、启动网关

    2、存储节点 连接到网关

    3、用户浏览器请求网关，请求和存储节点交换信令

    4、信令交换完成，使用dataChannel进行通信

    五、网关设计

    网关主要是一个websocket server 采用golang编写

    网关设计主要分成了几大模块

    1、room

    room是一个公共领域，主要是做client和node的鉴权操作，如果鉴权成功，那么会进入Manager进行调度

    2、抽象client和node

    client和node 有一个共性是他们都具有连接属性，所以应该设计一个公共接口，把他们的共性抽象出来

    1. type Lifecycle interface {
    2. 	Stop()
    3. 	Start()
    4. }
    5.  
    6. type Describable interface {
    7. 	Name() string
    8. 	Type() ConnType
    9. }
    10.  
    11.  
    12. type Component interface {
    13. 	Lifecycle
    14. 	Describable
    15. 	AddCallback(ctx EventCallback)
    16. 	GetConnNumber() uint64
    17. 	SetContext(interface{})
    18. 	GetContext() interface{}
    19. 	Product(ctx Context)
    20. }

    通过添加一个AddCallback回调函数，确保不同模块的业务处理完全玻璃开，node的逻辑就在node里处理，client的逻辑只在client里处理，不能把不同模块的代码交叉处理，至于上下问的传输，统一抽象一个Context，里面抽象存储着我们需要的上下文信息，供给不同的回调函数以及协成之间传输使用。

    1. type Context interface {
    2. 	GetData() []byte
    3. 	Error() error
    4. 	MessageType() int
    5. 	GetContext() interface{}
    6. 	SetContext(interface{})
    7. 	Name() string
    8. }
    9.  
    10. type EventCallback interface {
    11. 	OnReceive(Context, *sync.WaitGroup)
    12. 	OnError(Context, *sync.WaitGroup)
    13. 	OnClose(Context, *sync.WaitGroup)
    14. 	OnWriteComplete(Context, *sync.WaitGroup)
    15. }
    16.  
    17. type NodeCallback interface {
    18. 	OnReceive(Context, *sync.WaitGroup)
    19. 	OnError(Context, *sync.WaitGroup)
    20. 	OnClose(Context, *sync.WaitGroup)
    21. 	OnWriteComplete(Context, *sync.WaitGroup)
    22. }

    4、manager

    负责整体调度，比如client进入manager后，查找当前可以用的node存储节点，找到后进行信令交换

    节点调度，遍历查找拥有client最少的node节点，然后进行通信

    1. // 选择最优线路
    2. func (m *Manager) selectNode() *node.NodeClient {
    3. 	if len(m.nodeTree) == 0 {
    4. 		return nil
    5. 	}
    6.  
    7. 	// 找一个挂载链接最少的节点
    8. 	var usableNode *node.NodeClient
    9. 	var weight uint64
    10. 	for _, conn := range m.nodeTree {
    11. 		if uint64(conn.GetConnNumber()) <= weight {
    12. 			usableNode = conn.(*node.NodeClient)
    13. 		}
    14. 	}
    15. 	return usableNode
    16. }

    六、node节点收到信令后进行应答

    1、创建RTCConnection

    1. blockSize := 16384
    2. 	//前端页面会对sdp进行base64的encode
    3. 	b, err := base64.StdEncoding.DecodeString(sdp)
    4. 	if err != nil {
    5. 		log.Error("error:%s", err)
    6. 		return nil
    7. 	}
    8.  
    9. 	str, err := url.QueryUnescape(string(b))
    10. 	if err != nil {
    11. 		log.Error("error:%s", err)
    12. 		return nil
    13. 	}
    14.  
    15. 	sdpDes := webrtc.SessionDescription{}
    16. 	fmt.Println(str)
    17. 	err = json.Unmarshal([]byte(str), &sdpDes)
    18. 	if err != nil {
    19. 		log.Error("json.Unmarshal err:%s", err)
    20. 		return nil
    21. 	}
    22.  
    23. 	//创建pc， 并且指定stun服务器
    24. 	pc, err := webrtc.NewPeerConnection(webrtc.Configuration{
    25. 		ICEServers: []webrtc.ICEServer{
    26. 			{
    27. 				URLs: []string{"stun:"},
    28. 			},
    29. 		},
    30. 	})
    31.  
    32. 	stat, err := os.Stat("/home/zhanglei/Downloads/《实践论》(原文)毛泽东.pdf")
    33. 	if err != nil {
    34. 		log.Error("os.Stat %s error:%s", path, err)
    35. 		return nil
    36. 	}
    37.  
    38. 	if offset > stat.Size() {
    39. 		log.Error("offset(%d) > stat.Size(%d)", offset, stat.Size())
    40. 		return nil
    41. 	}
    42.  
    43. 	chunkSize := int(math.Ceil(float64(stat.Size() / int64(blockSize))))
    44. 	currentChunkSize := int(math.Ceil(float64(offset / int64(blockSize))))
    45.  
    46. 	if err != nil {
    47. 		log.Error("%s", err)
    48. 		return nil
    49. 	}
    50.  
    51. 	pc.OnConnectionStateChange(func(s webrtc.PeerConnectionState) {
    52. 		fmt.Println("OnConnectionStateChange")
    53. 		fmt.Printf("Peer Connection State has changed: %s (answerer)\n", s.String())
    54. 		if s == webrtc.PeerConnectionStateFailed {
    55. 			fmt.Println("webrtc.PeerConnectionStateFailed")
    56. 		}
    57. 	})
    58.  
    59. 	// Register data channel creation handling
    60. 	pc.OnDataChannel(func(d *webrtc.DataChannel) {
    61. 		fmt.Printf("New DataChannel %s %d\n", d.Label(), d.ID())
    62.  
    63. 		// Register channel opening handling
    64. 		d.OnOpen(func() {
    65. 			fmt.Printf("Data channel '%s'-'%d' open. Random messages will now be sent to any connected DataChannels every 5 seconds\n", d.Label(), d.ID())
    66.  
    67. 			stat, err := os.Stat("/home/zhanglei/Downloads/《实践论》(原文)毛泽东.pdf")
    68. 			chunkSize = int(math.Ceil(float64(stat.Size() / int64(blockSize))))
    69.  
    70. 			// 握手
    71. 			var chunk ChunkMessage
    72. 			chunk.Class = HANDSHAKE
    73. 			chunk.ChunkSize = uint64(chunkSize)
    74. 			handShakeBytes := serialize(&chunk)
    75.  
    76. 			err = d.Send(handShakeBytes.Bytes())
    77. 			if err != nil {
    78. 				log.Error("%s", err)
    79. 				return
    80. 			}
    81. 		})
    82.  
    83. 		// Register text message handling
    84. 		d.OnMessage(func(msg webrtc.DataChannelMessage) {
    85. 			fmt.Printf("Message from DataChannel '%s': '%s'\n", d.Label(), string(msg.Data))
    86. 			data, err := unSerialize(msg.Data)
    87. 			if err != nil {
    88. 				log.Error("os.Open is : %s", err)
    89. 				return
    90. 			}
    91.  
    92. 			if data.Class == ACK {
    93. 				handle, err := os.Open("/home/zhanglei/Downloads/《实践论》(原文)毛泽东.pdf")
    94. 				if err != nil {
    95. 					log.Error("os.Open is : %s", err)
    96. 					return
    97. 				}
    98.  
    99. 				defer handle.Close()
    100.  
    101. 				bufferBytes := make([]byte, blockSize)
    102. 				read, err := handle.Read(bufferBytes)
    103. 				if err != nil {
    104. 					log.Error("handle.Read is : %s", err)
    105. 					return
    106. 				}
    107.  
    108. 				if read < blockSize {
    109. 					bufferBytes = bufferBytes[:read]
    110. 				}
    111.  
    112. 				var chunk ChunkMessage
    113. 				chunk.Class = SEND
    114. 				chunk.ChunkSize = uint64(chunkSize)
    115. 				chunk.CurrentChunk = uint64(currentChunkSize)
    116. 				chunk.Data = bufferBytes
    117.  
    118. 				// 打包发送
    119. 				err = d.Send(serialize(&chunk).Bytes())
    120. 				if err != nil {
    121. 					log.Error("%s", err)
    122. 					return
    123. 				}
    124. 				return
    125. 			}
    126.  
    127. 			if data.Class == RECEIVE {
    128. 				handle, err := os.Open("/home/zhanglei/Downloads/《实践论》(原文)毛泽东.pdf")
    129. 				if err != nil {
    130. 					log.Error("os.Open is : %s", err)
    131. 					return
    132. 				}
    133.  
    134. 				if data.CurrentChunk == uint64(chunkSize) {
    135. 					log.Info("data transfer finish")
    136. 					return
    137. 				}
    138.  
    139. 				nextChunk := data.CurrentChunk + 1
    140.  
    141. 				bytes := make([]byte, blockSize)
    142. 				read, err := handle.ReadAt(bytes, int64(nextChunk)*int64(blockSize))
    143.  
    144. 				if err != nil {
    145. 					if !errors.Is(err, io.EOF) {
    146. 						log.Error("handle.Read is : %s", err)
    147. 						return
    148. 					}
    149.  
    150. 				}
    151.  
    152. 				if read < blockSize {
    153. 					bytes = bytes[:read]
    154. 				}
    155.  
    156. 				var sendData ChunkMessage
    157. 				sendData.Class = SEND
    158. 				sendData.CurrentChunk = nextChunk
    159. 				sendData.Data = bytes
    160. 				sendData.ChunkSize = uint64(chunkSize)
    161. 				sendDataBytes := serialize(&sendData)
    162. 				log.Info(" read %d", nextChunk)
    163.  
    164. 				err = d.Send(sendDataBytes.Bytes())
    165. 				if err != nil {
    166. 					log.Error("%s", err)
    167. 					return
    168. 				}
    169.  
    170. 				// 最后一块
    171. 				if nextChunk == uint64(chunkSize) {
    172. 					d.Close()
    173. 					pc.Close()
    174. 				}
    175. 				return
    176. 			}
    177.  
    178. 		})
    179. 	})
    180.  
    181. 	_, err = pc.CreateDataChannel("sendDataChannel", nil)
    182. 	if err != nil {
    183. 		log.Error("error:%s", err)
    184. 		return nil
    185. 	}
    186.  
    187. 	//channel.OnOpen(func() {
    188. 	//
    189. 	//})
    190.  
    191. 	//设置远端的sdp
    192. 	if err = pc.SetRemoteDescription(sdpDes); err != nil {
    193. 		log.Error("error:%s", err)
    194. 		return nil
    195. 	}
    196.  
    197. 	//创建协商结果
    198. 	answer, err := pc.CreateAnswer(nil)
    199. 	if err != nil {
    200. 		log.Error("error:%s", err)
    201. 		return nil
    202. 	}
    203.  
    204. 	pc.OnICECandidate(func(i *webrtc.ICECandidate) {
    205. 		fmt.Println("OnICECandidate")
    206. 		fmt.Println(i)
    207.  
    208. 	})
    209.  
    210. 	err = pc.SetLocalDescription(answer)
    211. 	if err != nil {
    212. 		log.Error("error:%s", err)
    213. 		return nil
    214. 	}
    215.  
    216. 	//等待ice结束
    217. 	gatherCmp := webrtc.GatheringCompletePromise(pc)
    218. 	<-gatherCmp
    219.  
    220. 	//将协商并且收集完candidate的answer，输出到控制台
    221. 	answerBytes, err := json.Marshal(*pc.LocalDescription())
    222. 	if err != nil {
    223. 		log.Error("error:%s", err)
    224. 		return nil
    225. 	}
    226.  
    227. 	pc.OnICECandidate(func(candidate *webrtc.ICECandidate) {
    228. 		fmt.Println("OnICECandidate")
    229. 		fmt.Println(candidate)
    230. 	})
    231.  
    232. 	t := &Transfer{
    233. 		sdp:              sdp,
    234. 		pc:               pc,
    235. 		offset:           uint64(offset),
    236. 		currentChunkSize: currentChunkSize,
    237. 		chunkSize:        chunkSize,
    238. 		answerSdp:        answerBytes,
    239. 		blockSize:        blockSize,
    240. 	}
    241. 	return t

    2、打包传输

    传输没有采用protobuf，自己写了个二进制传输

    1. func serialize(data *ChunkMessage) *bytes.Buffer {
    2. 	data.Version = CodeVersion
    3. 	writeBuffer := bytes.NewBuffer(nil)
    4. 	writeBuffer.Write([]byte{data.Version})
    5. 	writeBuffer.Write([]byte{data.Class})
    6.  
    7. 	// ChunkSize
    8. 	binary.Write(writeBuffer, binary.BigEndian, data.ChunkSize)
    9.  
    10. 	// CurrentChunk
    11. 	binary.Write(writeBuffer, binary.BigEndian, data.CurrentChunk)
    12.  
    13. 	// DataLen
    14. 	data.DataLen = uint64(len(data.Data))
    15. 	binary.Write(writeBuffer, binary.BigEndian, data.DataLen)
    16.  
    17. 	if len(data.Data) > 0 {
    18. 		// 添加body
    19. 		writeBuffer.Write(data.Data)
    20. 	}
    21.  
    22. 	return writeBuffer
    23. }
    24.  
    25. //判断我们系统中的字节序类型
    26. func systemEdian() binary.ByteOrder {
    27. 	var i int = 0x1
    28. 	bs := (*[int(unsafe.Sizeof(0))]byte)(unsafe.Pointer(&i))
    29. 	if bs[0] == 0 {
    30. 		return binary.LittleEndian
    31. 	} else {
    32. 		return binary.BigEndian
    33. 	}
    34. }
    35.  
    36. func unSerialize(data []byte) (*ChunkMessage, error) {
    37. 	buf := bytes.NewBuffer(data)
    38. 	fmt.Println(buf)
    39. 	var chunk ChunkMessage
    40. 	binary.Read(buf, systemEdian(), &chunk.Version)
    41. 	binary.Read(buf, systemEdian(), &chunk.Class)
    42. 	binary.Read(buf, systemEdian(), &chunk.ChunkSize)
    43. 	binary.Read(buf, systemEdian(), &chunk.CurrentChunk)
    44. 	binary.Read(buf, systemEdian(), &chunk.DataLen)
    45. 	//chunkSize := uint64(unsafe.Pointer(&buf.Bytes()))
    46. 	//chunk.ChunkSize = chunkSize
    47. 	return &chunk, nil
    48. }

    7、js前端webrtc提供报价

    创建webrtc连接

    1. var pcConfig = {
    2.             'iceServers': [{
    3.                 'urls': 'stun:',
    4.             }]
    5.         };
    6.         localConnection = new RTCPeerConnection(pcConfig);
    7.  
    8.         receiveDataChannel = localConnection.createDataChannel("receiveDataChannel")
    9.  
    10.         receiveDataChannel.binaryType = "arraybuffer"
    11.  
    12.         receiveDataChannel.addEventListener('open', dataChannel.onopen);
    13.         receiveDataChannel.addEventListener('close', dataChannel.onclose);
    14.         receiveDataChannel.addEventListener('message', dataChannel.onmessage);
    15.         receiveDataChannel.addEventListener('error', dataChannel.onError);
    16.  
    17.         try {
    18.             this.offer = await localConnection.createOffer();
    19.         } catch (e) {
    20.             console.log('Failed to create session description: ', e);
    21.             return
    22.         }
    23.  
    24.         try {
    25.             await localConnection.setLocalDescription(this.offer)
    26.         } catch (e) {
    27.             console.log('Failed to create session description: ', e);
    28.             return
    29.         }
    30.  
    31.         //eyJ1c2VyX3V1aWQiOiI1ZmZkNDE0N2JkMTMyNWNmMjYwNDAyMWYwODA5OWUyMyIsImxvZ2luX3RpbWUiOjE2Njg0NzgzOTEsIm5vd190aW1lIjoxNjY5OTgzNzkwLCJyYW5fc3RyIjoiZDA3MTczNzI3NjFjMzY0MGU2NmRlYWI5YmYyODZhNzYiLCJzaWduIjoiZjc3NzI0YjZmMTc3MzczNmVhZWFkMTM2NzllNTE0NTcifQ==
    32.  
    33.         transfer.ws.send((JSON.stringify(downloadRequest)));

    前端对收到的数据进行序列化和反序列化

    1. function serialize(data) {
    2.     var bufLen = protoColMinSize;
    3.     if (!data.Data) {
    4.         bufLen += 0;
    5.     } else {
    6.         bufLen += data.Data.length;
    7.     }
    8.     data.Version = 1;
    9.     var protocolBuf = new ArrayBuffer(bufLen);
    10.     const bufView = new DataView(protocolBuf);
    11.     bufView.setUint8(0, data.Version);
    12.  
    13.     bufView.setUint8(1, data.Class);
    14.  
    15.     if (!data.ChunkSize) {
    16.         data.ChunkSize = 0
    17.     }
    18.     bufView.setBigUint64(2, BigInt(data.ChunkSize));
    19.  
    20.  
    21.     if (!data.CurrentChunk) {
    22.         data.CurrentChunk = 0
    23.     }
    24.     bufView.setBigUint64(10, BigInt(data.CurrentChunk));
    25.  
    26.     if (data.Data && data.Data.length > 0) {
    27.         bufView.setBigUint64(18, BigInt(data.Data.length));
    28.     } else {
    29.         bufView.setBigUint64(18, BigInt(0));
    30.     }
    31.  
    32.     console.log(protocolBuf)
    33.     return protocolBuf;
    34. }
    35.  
    36. function unSerialize(bytes) {
    37.  
    38.     var versionView = new DataView(bytes).getUint8(0);
    39.     // 最小长度
    40.     var classByteView = new DataView(bytes).getUint8(1);
    41.     // chunk 长度
    42.     var chunkSizeView = parseInt(new DataView(bytes).getBigUint64(2));
    43.     var currentChunkView = parseInt(new DataView(bytes).getBigUint64(10));
    44.     var bodyLenView = parseInt(new DataView(bytes).getBigUint64(18));
    45.     var returnData = {
    46.         Version: versionView,
    47.         Class: classByteView,
    48.         ChunkSize: (chunkSizeView),
    49.         CurrentChunk: currentChunkView,
    50.         PayloadLength: bodyLenView,
    51.         Payload: [],
    52.     };
    53.  
    54.     if (bodyLenView > 0) {
    55.         returnData.Payload = new Uint8Array(bytes, protoColMinSize, bodyLenView)
    56.     }
    57.  
    58.     return returnData;
    59. }

    八、webrtc文件传输的优缺

    1）优点

    1、点对点传输，不经过中继服务器

    2、民用带宽比较便宜，最差的情况下是带宽打满，不会出现很高的流量费用

    3、可以自建存储，存储节点可以使用群辉，可以买自己服务器

    4、不需要固定的ip地址

    2）缺点

    1、宽带线民用的不知道能申请几根

    2、存储维护硬件也是一个麻烦的事情，硬盘很可能出现故障，运维也是一个头痛的事情

  • 相关阅读:
    Patroni的执行流
    Chapter 2 Gradient Descent
    HTML5期末考核大作业——学生网页设计作业源码HTML+CSS+JavaScript 中华美德6页面带音乐文化
    【Java 进阶篇】数据定义语言（DDL）详解
    如何使用现有工具三分钟之内从无到有设计一款滤波器？
    JAVA 7-3 统计投票
    java中的并发工具类
    JAVA8时间工具类
    iOS 16适配屏幕旋转强制转屏切换大总结
    Lua篇笔记
  • 原文地址：https://blog.csdn.net/qq_32783703/article/details/128168660