點擊上方“LiveVideoStack”關注我們

作者 | 王朋闖

本文為王朋闖老師創作的系列ion文章，LiveVideoStack已獲得授權發布，未來將持續更新。

一、為什么用ion-sfu

1.簡介

ion-sfu作為ion分布式架構里的核心模塊，SFU是選擇轉發單元的簡稱，可以分發WebRTC的媒體流。ion-sfu從pion/ion拆分出來，經過社區打磨，是目前GO方案中最成熟且使用最廣的SFU。

https://github.com/pion/ion

已經有多家開始商用了，這點國外公司比較快，比如：100ms、Screenleap和Tandem等。

100ms：https://www.100ms.live/

Screenleap：https://www.screenleap.com/

Tandem：https://tandem.chat/

2.ion-sfu優點

?

• 純GO，開發效率高，且能幫你繞過很多坑

• 單進程多協程模型：

  - 可以利用多核

  -?大大降低級聯/單端口復雜度（其他SFU，可能存在本機不同worker間relay的問題；監聽單端口時，存在worker間搶包的問題）

• 高并發，曾在谷歌云4核壓測到單房間50方會議 (大概2500路流-0.5Mbps)

• 功能全面：

  - 雙PeerConnection+多Track設計，有良好的瀏覽器兼容性，節省系統資源

  -?支持多對多音視頻通信

  -?支持大小流Simulcast

  -?支持屏幕分享Screenshare

  -?支持發言方自動檢測Audio-Level-Detect

  -?支持定制DataChannel

  -?支持節點間Relay

  -?支持單端口，大大降低部署難度

  -?完善的抗弱網機制，抗丟包40%左右，支持TWCC/REMB+PLI/FIR+Nack+SR/RR等

• 配套SDK完善，JS/Flutter/GO等

3.使用方式

ion-sfu使用方式有兩種：

• 作為服務使用，比如編譯帶grpc或jsonrpc信令的ion-sfu，然后再做一個自己的信令服務（推薦ion分布式套裝），遠程調用即可。

• 作為包使用，import導入，然后做二次開發。此時拋棄了cmd下邊的信令層，只需導入pkg/sfu下邊的包即可，然后自行定制信令層，可以在sfu、session、peer層面，通過繼承接口定制自己的業務，比較復雜。

import (sfu "github.com/pion/ion-sfu/pkg/sfu" )

二、架構與模塊

上面給一個簡單架構圖，很多細節表示不出來，需要看代碼。

1.簡介

得益于GO，ion-sfu整體代碼精簡，擁有極高的開發效率。結合現有SDK使用，可以避免很多坑：ion-sdk-js等。

ion-sfu基于pion/webrtc，所以代碼風格偏標準webrtc，比如：PeerConnection。因為是使用了標準API，熟悉了之后很容易看懂其他工程，比如：ion-sdk-go/js/flutter。

這樣從前到后，整體門檻都降低了。

2.工程組織

這里給出主要模塊列表：

├── Makefile //用來編譯二進制和grpc文件 ├── bin //編譯好的二進制目錄 ├── cmd │ └── signal //包含三個主文件 grpc、jsonrpc、allrpc ├── config.toml //配置文件 ├── examples //網頁示例目錄 ├── pkg├── buffer //buffer包，用于緩存包├── logger //日志├── middlewares //中間件，主要是支持自定義datachannel├── relay //中繼├── sfu //sfu主模塊，包含router、session、peer等├── stats //狀態統計└── twcc //transport-cc

3.信令層

信令代碼和主程序在一起，在cmd/signal/下邊。

• 支持jsonrpc，主要處理邏輯在：

func (p *JSONSignal) Handle(ctx context.Context, conn *jsonrpc2.Conn, req *jsonrpc2.Request) {

• 支持grpc，主要處理邏輯在：

func (s *SFUServer) Signal(stream pb.SFU_SignalServer) error {

而allrpc，是jsonrpc和grpc的合體封裝，運行時會進入上面兩個函數。

信令很簡單：

• join：加入一個session。

• description：發起offer或回復answer，用于協商和重協商。

• trickle：發送trickle candidate。

另外，出于簡單考慮，一些信令和事件，直接走datachannel了，比如：大小流切換、聲音檢測、自定義信令等。

4.媒體層

媒體層的主要模塊：

├── audioobserver.go //聲音檢測 ├── datachannel.go //dc中間件的封裝 ├── downtrack.go //下行track ├── helpers.go //工具函數集 ├── mediaengine.go //SDP相關codec、rtp參數設置 ├── peer.go //peer封裝，一個peer包含一個publisher和一個subscriber，雙pc設計 ├── publisher.go //publisher，封裝上行pc ├── receiver.go //subscriber，封裝下行pc ├── router.go //router，包含pc、session、一組receivers ├── sequencer.go //記錄包的信息：序列號sn、偏移、時間戳ts等 ├── session.go //會話，包含多個peer、dc ├── sfu.go //分發單元，包含多個session、dc ├── simulcast.go //大小流配置 ├── subscriber.go //subscriber，封裝下行pc、DownTrack、dc └── turn.go //內置turn server

相比以前版本，增加了一些interface，主要是為了作為包使用時，封裝自己的類。

三、主函數與信令流程

1.主函數

這里拿jsonrpc來分析，其他rpc流程上是一樣的。

func main() {if !parse() {showHelp()os.Exit(-1)}//創建SFU和DC，這里的DC用于Simulcast和AudioLevels := sfu.NewSFU(conf)dc := s.NewDatachannel(sfu.APIChannelLabel)dc.Use(datachannel.SubscriberAPI)//接下來是標準websocket服務器啟動的流程upgrader := websocket.Upgrader{CheckOrigin: func(r *http.Request) bool {return true},ReadBufferSize: 1024,WriteBufferSize: 1024,}//這里jsonrpc基于websocket，websocket從標準http upgrade過來的http.Handle("/ws", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {c, err := upgrader.Upgrade(w, r, nil)if err != nil {panic(err)}defer c.Close()//這里創建了JSONSignal，每次真實請求到來時會新建一個Peer，進入Handle函數處理p := server.NewJSONSignal(sfu.NewPeer(s), logger)defer p.Close()jc := jsonrpc2.NewConn(r.Context(), websocketjsonrpc2.NewObjectStream(c), p)<-< span="">jc.DisconnectNotify()}))go startMetrics(metricsAddr)var err errorif key != "" && cert != "" {logger.Info("Started listening", "addr", "https://"+addr)err = http.ListenAndServeTLS(addr, cert, key, nil)} else {logger.Info("Started listening", "addr", "http://"+addr)err = http.ListenAndServe(addr, nil)}if err != nil {panic(err)} }

2.協商&重協商

協商(negotiate)：

WebRTC對外的類是PeerConnection，簡稱PC，通過信令服務交換SDP給PC進行操作。協商就是指雙方通過信令交換SDP，通過PC的一些接口，達到協商雙方的媒體格式、傳輸地址端口等信息，從而實現推流和播放的目的。

一次協商完整流程：

本端CreateOffer-》本端SetLocalDescription(offer)-》本端發送offer-》對端SetRemoteDescription(offer)-》對端CreateAnswer-》SetLocalDescription(answer)-》對端對端返回answer-》本端SetRemoteDescription(answer)

重協商(renegotiate)：

就是指再次協商。

為什么要重協商？

因為客戶端和服務器的track都是變化的，重協商是通知對端的必要手段，比如：客戶端發起屏幕分享，服務器有人進出房間等。

重協商的原則：

誰變化誰發起（offer)。

3.信令流程

首先，客戶端ws連接成功：

服務端會建立一個Peer，可以參考上邊代碼。

然后，客戶端發起第一次協商：

客戶端pc.CreateOffer（一個只包含dc的offer）-》pc.SetLocalDescription(offer)，然后把offer放入Join信令，發送給服務端，然后服務器協商【pc.SetRemoteDescription(offer)-》pc.CreateAnswer-》pc.SetLocalDescription(answer)】，返回answer給客戶端，至此完成數據通道（datachannel)建立。首先打通dc，是為了方便audio-level/simulcast通道的建立，此時也可以創建自定dc做定制業務。

接下來，服務端發起第二次協商：

服務端pc.CreateOffer，SetLocalDescription，發送offer，此時offer會攜帶上此房間內的所有track信息，客戶端收到后會CreateAnswer，SetLocalDescription，把answer返回來，然后服務端pc.SetRemoteDescription(answer)，此時客戶端可以收到服務器此房間內的所有流了。

最后，客戶端publish發流時會發起第三次協商：

同第一次流程一樣，不同的是同時攜帶了音視頻的track，本次協商完成后，服務器可以收到客戶端的流了，收到之后會對同房間內的其他客戶端發起重協商。

往后只要客戶端或服務器track有變化，都會再次發起重協商。

4.代碼分析

JsonRPC所有的信令都會進入Handle函數。為了簡化流程，可以暫時不看Trickle和OnIceCandidate函數，這個是開啟trickle-ICE時才會有。

// Handle incoming RPC call events like join, answer, offer and trickle // JSONSignal是繼承了LocalPeer，所以會繼承一些屬性和回調：OnOffer等。 // 可以在瀏覽器端ws網絡工具里查看具體信令內容 func (p *JSONSignal) Handle(ctx context.Context, conn *jsonrpc2.Conn, req *jsonrpc2.Request) {replyError := func(err error) {_ = conn.ReplyWithError(ctx, req.ID, &jsonrpc2.Error{Code: 500,Message: fmt.Sprintf("%s", err),})}switch req.Method {case "join":// 首先客戶端會發join信令過來var join Joinerr := json.Unmarshal(*req.Params, &join)if err != nil {p.Logger.Error(err, "connect: error parsing offer")replyError(err)break}//設置OnOffer，即SFU發起offer時（重協商），會使用這個回調，比如重協商時，因為有很多客戶端peer同時連到SFU，每個Peer的Track增刪時，SFU需要向其他Peer重協商來告訴Track的變更p.OnOffer = func(offer *webrtc.SessionDescription) {if err := conn.Notify(ctx, "offer", offer); err != nil {p.Logger.Error(err, "error sending offer")}}//設置OnIceCandidate，即SFU在ICE流程獲取到新候選時，會回調這個函數，告訴客戶端新增了啥候選p.OnIceCandidate = func(candidate *webrtc.ICECandidateInit, target int) {if err := conn.Notify(ctx, "trickle", Trickle{Candidate: *candidate,Target: target,}); err != nil {p.Logger.Error(err, "error sending ice candidate")}}//加入某個會話(房間)err = p.Join(join.SID, join.UID, join.Config)if err != nil {replyError(err)break}//根據offer回復answeranswer, err := p.Answer(join.Offer)if err != nil {replyError(err)break}_ = conn.Reply(ctx, req.ID, answer)//如果是客戶端發offer，回復answer，此時為客戶端發起重協商case "offer":var negotiation Negotiationerr := json.Unmarshal(*req.Params, &negotiation)if err != nil {p.Logger.Error(err, "connect: error parsing offer")replyError(err)break}answer, err := p.Answer(negotiation.Desc)if err != nil {replyError(err)break}_ = conn.Reply(ctx, req.ID, answer)//如果是客戶端發answer，設置SetRemoteDescription即可case "answer":var negotiation Negotiationerr := json.Unmarshal(*req.Params, &negotiation)if err != nil {p.Logger.Error(err, "connect: error parsing offer")replyError(err)break}err = p.SetRemoteDescription(negotiation.Desc)if err != nil {replyError(err)}//如果是客戶端發送Trickle-ICE的候選過來，設置即可case "trickle":var trickle Trickleerr := json.Unmarshal(*req.Params, &trickle)if err != nil {p.Logger.Error(err, "connect: error parsing candidate")replyError(err)break}err = p.Trickle(trickle.Candidate, trickle.Target)if err != nil {replyError(err)}} }

注意OnOffer是服務器重協商的回調，即房間內某客戶端track有變化，服務器會回調此函數通知其他客戶端。

總結一句話，客戶端《---》SFU的核心邏輯就是不斷重協商，誰變化誰發起offer。

---

作者簡介：

王朋闖：前百度RTN資深工程師，前金山云RTC技術專家，前VIPKID流媒體架構師，ION開源項目發起人。

特別說明：

本文發布于知乎，已獲得作者授權轉載。

---

掃描圖中二維碼或點擊閱讀原文

了解大會更多信息

喜歡我們的內容就點個“在看”吧！

超強干貨來襲 云風專訪：近40年碼齡，通宵達旦的技術人生

總結

以上是生活随笔為你收集整理的大话ion系列（一）的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。