C#開源實(shí)現(xiàn)MJPEG流傳輸

作者：Dennis Gao 2013-08-21 13:34:52

目前已經(jīng)實(shí)現(xiàn)了UDP+RTP 方式在不同物理機(jī)之間的媒體流傳輸。當(dāng)然，由于沒有基于 .NET 的媒體流壓縮實(shí)現(xiàn)，所以直接傳輸?shù)穆銏DBitmap。不過(guò)要求不高，幀率低一些，機(jī)器性能強(qiáng)一些，看著也很流暢。

許久以前寫了篇文章《基于.NET打造IP智能網(wǎng)絡(luò)視頻監(jiān)控系統(tǒng)》，記錄和介紹了自己幾年來(lái)積累和演練的一個(gè)系統(tǒng)。發(fā)現(xiàn)幾個(gè)月過(guò)去了，沒有任何進(jìn)展。

能在桌面客戶端上看到視頻圖像的功能已經(jīng)完成了。下面需要考慮，如何通過(guò)瀏覽器來(lái)查看視頻。

在不考慮使用 Flash、ActiveX 的條件下，貌似只能選擇 MJPEG 方式。目前還沒有研究在 HTML5 下視頻是如何處理的，以后有時(shí)間可以探索。

什么是 MJPEG？

看這里：

http://en.wikipedia.org/wiki/Motion_JPEG
http://zh.wikipedia.org/wiki/Motion_JPEG
http://baike.baidu.com/view/2098077.htm

當(dāng)然，我主要關(guān)注 MJPEG over HTTP 這段。

M-JPEG over HTTP
HTTP streaming separates each image into individual HTTP replies on a specified marker. RTP streaming creates packets of a sequence of JPEG images that can be received by clients such as QuickTime or VLC.
In response to a GET request for a MJPEG file or stream, the server streams the sequence of JPEG frames over HTTP. A special mime-type content type multipart/x-mixed-replace;boundary=<boundary-name> informs the client to expect several parts (frames) as an answer delimited by <boundary-name>. This boundary name is expressly disclosed within the MIME-type declaration itself. The TCP connection is not closed as long as the client wants to receive new frames and the server wants to provide new frames. Two basic implementations of a M-JPEG streaming server are cambozola and MJPG-Streamer. The more robust ffmpeg-server also provides M-JPEG streaming support.

也就是說(shuō)，建立 HTTP 連接后，服務(wù)端在 Response 消息中先發(fā)一個(gè)數(shù)據(jù)頭 Header 告訴客戶端，我后面的都是 JPEG 圖片。圖片之間使用 boundary-name 來(lái)區(qū)分，每個(gè)圖片前都有自己的數(shù)據(jù)頭來(lái)描述圖片數(shù)據(jù)長(zhǎng)度。

MJPEG數(shù)據(jù)頭定義

/// <summary>  
    /// 流頭部  
    /// </summary>  
    public string StreamHeader  
    {  
      get 
      {  
        return "HTTP/1.1 200 OK" +  
               "\r\n" +  
               "Content-Type: multipart/x-mixed-replace; boundary=" + this.Boundary +  
               "\r\n";  
      }  
    }

/// <summary>  
    /// 圖片頭部  
    /// </summary>  
    public string PayloadHeader  
    {  
      get 
      {  
        return "\r\n" +  
               this.Boundary +  
               "\r\n" +  
               "Content-Type: image/jpeg" +  
               "\r\n" +  
               "Content-Length: " + _contentLengthString +  
               "\r\n\r\n";  
      }  
    }

這里的 Boundary 可以是任意字符串，只要你覺得唯一并能區(qū)分即可，比如我可以設(shè)置為“--dennisgao”。

#p#

服務(wù)器端實(shí)現(xiàn)

Http 服務(wù)器其實(shí)就是個(gè)支持 Tcp 連接的服務(wù)器。

private AsyncTcpServer _server;  
 
_server = new AsyncTcpServer(Port);  
_server.Encoding = Encoding.ASCII;

public void Start()  
    {  
      _server.Start(10);  
      _server.ClientConnected += new EventHandler<TcpClientConnectedEventArgs>(OnClientConnected);  
      _server.ClientDisconnected += new EventHandler<TcpClientDisconnectedEventArgs>(OnClientDisconnected);  
    }  
 
    public void Stop()  
    {  
      _server.Stop();  
      _server.ClientConnected -= new EventHandler<TcpClientConnectedEventArgs>(OnClientConnected);  
      _server.ClientDisconnected -= new EventHandler<TcpClientDisconnectedEventArgs>(OnClientDisconnected);  
    }  
 
    private void OnClientConnected(object sender, TcpClientConnectedEventArgs e)  
    {  
      _clients.AddOrUpdate(e.TcpClient.Client.RemoteEndPoint.ToString(), e.TcpClient, (n, o) => { return e.TcpClient; });  
    }  
 
    private void OnClientDisconnected(object sender, TcpClientDisconnectedEventArgs e)  
    {  
      TcpClient clientToBeThrowAway;  
      _clients.TryRemove(e.TcpClient.Client.RemoteEndPoint.ToString(), out clientToBeThrowAway);  
    }

這里可以參考兩篇文章中的實(shí)現(xiàn)。

C#異步TCP服務(wù)器完整實(shí)現(xiàn)
C#對(duì)TCP客戶端的狀態(tài)封裝

發(fā)送圖片數(shù)據(jù)

首先要保證，對(duì)一個(gè)HTTP連接只能發(fā)一次流頭，因?yàn)楹竺媸墙舆B不斷的圖片數(shù)據(jù)。當(dāng)然，發(fā)點(diǎn)別的數(shù)據(jù)客戶端也不會(huì)解碼。

private void WriteStreamHeader()  
    {  
      if (_clients.Count > 0)  
      {  
        foreach (var item in _clients)  
        {  
          Logger.Debug(string.Format(CultureInfo.InvariantCulture,  
            "Writing stream header, {0}, {1}{2}", item.Key, Environment.NewLine, StreamHeader));  
 
          _server.SyncSend(item.Value, StreamHeader);  
 
          TcpClient clientToBeThrowAway;  
          _clients.TryRemove(item.Key, out clientToBeThrowAway);  
        }  
      }  
    }

發(fā)送圖片數(shù)據(jù)時(shí)，要保證圖片的前面是圖片頭和長(zhǎng)度信息，數(shù)據(jù)尾部要有換行符。

private void WritePayload(byte[] payload)  
    {  
      string payloadHeader = this.PayloadHeader.Replace(_contentLengthString, payload.Length.ToString());  
      string payloadTail = "\r\n";  
 
      Logger.Debug(string.Format(CultureInfo.InvariantCulture,  
        "Writing payload header, {0}{1}", Environment.NewLine, payloadHeader));  
 
      byte[] payloadHeaderBytes = _server.Encoding.GetBytes(payloadHeader);  
      byte[] payloadTailBytes = _server.Encoding.GetBytes(payloadTail);  
      byte[] packet = new byte[payloadHeaderBytes.Length + payload.Length + payloadTail.Length];  
      Buffer.BlockCopy(payloadHeaderBytes, 0, packet, 0, payloadHeaderBytes.Length);  
      Buffer.BlockCopy(payload, 0, packet, payloadHeaderBytes.Length, payload.Length);  
      Buffer.BlockCopy(payloadTailBytes, 0, packet, payloadHeaderBytes.Length + payload.Length, payloadTailBytes.Length);  
 
      _server.SendToAll(packet);  
    }

結(jié)果演示

在可以成功發(fā)送流信息和圖片信息后，就可以在瀏覽器上查看視頻了。當(dāng)然，我用的 Google Chrome 。IE10 好奇葩，它會(huì)把流當(dāng)成文件不停的下載，搞不懂。

遠(yuǎn)程訪問(wèn)

局域網(wǎng)內(nèi)的無(wú)線設(shè)備，只要瀏覽器支持 MJPEG ，均可以查看視頻。我測(cè)試了 iPad 上的 Safari 是可以的，但 Chrome 卻直接解析成亂碼。

當(dāng)然，如果在路由器上配置轉(zhuǎn)發(fā)規(guī)則，就可以在外網(wǎng)訪問(wèn)了。

#p#

完整代碼

public class MJpegStreamingServer  
  {  
    private static string _contentLengthString = "__PayloadHeaderContentLength__";  
    private AsyncTcpServer _server;  
    private ConcurrentDictionary<string, TcpClient> _clients;  
 
    public MJpegStreamingServer(int listenPort)  
      : this(listenPort, "--dennisgao")  
    {  
    }  
 
    public MJpegStreamingServer(int listenPort, string boundary)  
    {  
      Port = listenPort;  
      Boundary = boundary;  
 
      _server = new AsyncTcpServer(Port);  
      _server.Encoding = Encoding.ASCII;  
      _clients = new ConcurrentDictionary<string, TcpClient>();  
    }  
 
    /// <summary>  
    /// 監(jiān)聽的端口  
    /// </summary>  
    public int Port { get; private set; }  
 
    /// <summary>  
    /// 分隔符  
    /// </summary>  
    public string Boundary { get; private set; }  
 
    /// <summary>  
    /// 流頭部  
    /// </summary>  
    public string StreamHeader  
    {  
      get 
      {  
        return "HTTP/1.1 200 OK" +  
               "\r\n" +  
               "Content-Type: multipart/x-mixed-replace; boundary=" + this.Boundary +  
               "\r\n";  
      }  
    }  
 
    /// <summary>  
    /// 圖片頭部  
    /// </summary>  
    public string PayloadHeader  
    {  
      get 
      {  
        return "\r\n" +  
               this.Boundary +  
               "\r\n" +  
               "Content-Type: image/jpeg" +  
               "\r\n" +  
               "Content-Length: " + _contentLengthString +  
               "\r\n\r\n";  
      }  
    }  
 
    public void Start()  
    {  
      _server.Start(10);  
      _server.ClientConnected += new EventHandler<TcpClientConnectedEventArgs>(OnClientConnected);  
      _server.ClientDisconnected += new EventHandler<TcpClientDisconnectedEventArgs>(OnClientDisconnected);  
    }  
 
    public void Stop()  
    {  
      _server.Stop();  
      _server.ClientConnected -= new EventHandler<TcpClientConnectedEventArgs>(OnClientConnected);  
      _server.ClientDisconnected -= new EventHandler<TcpClientDisconnectedEventArgs>(OnClientDisconnected);  
    }  
 
    private void OnClientConnected(object sender, TcpClientConnectedEventArgs e)  
    {  
      _clients.AddOrUpdate(e.TcpClient.Client.RemoteEndPoint.ToString(), e.TcpClient, (n, o) => { return e.TcpClient; });  
    }  
 
    private void OnClientDisconnected(object sender, TcpClientDisconnectedEventArgs e)  
    {  
      TcpClient clientToBeThrowAway;  
      _clients.TryRemove(e.TcpClient.Client.RemoteEndPoint.ToString(), out clientToBeThrowAway);  
    }  
 
    public void Write(Image image)  
    {  
      if (_server.IsRunning)  
      {  
        byte[] payload = BytesOf(image);  
 
        WriteStreamHeader();  
        WritePayload(payload);  
      }  
    }  
 
    private void WriteStreamHeader()  
    {  
      if (_clients.Count > 0)  
      {  
        foreach (var item in _clients)  
        {  
          Logger.Debug(string.Format(CultureInfo.InvariantCulture,  
            "Writing stream header, {0}, {1}{2}", item.Key, Environment.NewLine, StreamHeader));  
 
          _server.SyncSend(item.Value, StreamHeader);  
 
          TcpClient clientToBeThrowAway;  
          _clients.TryRemove(item.Key, out clientToBeThrowAway);  
        }  
      }  
    }  
 
    private void WritePayload(byte[] payload)  
    {  
      string payloadHeader = this.PayloadHeader.Replace(_contentLengthString, payload.Length.ToString());  
      string payloadTail = "\r\n";  
 
      Logger.Debug(string.Format(CultureInfo.InvariantCulture,  
        "Writing payload header, {0}{1}", Environment.NewLine, payloadHeader));  
 
      byte[] payloadHeaderBytes = _server.Encoding.GetBytes(payloadHeader);  
      byte[] payloadTailBytes = _server.Encoding.GetBytes(payloadTail);  
      byte[] packet = new byte[payloadHeaderBytes.Length + payload.Length + payloadTail.Length];  
      Buffer.BlockCopy(payloadHeaderBytes, 0, packet, 0, payloadHeaderBytes.Length);  
      Buffer.BlockCopy(payload, 0, packet, payloadHeaderBytes.Length, payload.Length);  
      Buffer.BlockCopy(payloadTailBytes, 0, packet, payloadHeaderBytes.Length + payload.Length, payloadTailBytes.Length);  
 
      _server.SendToAll(packet);  
    }  
 
    private byte[] BytesOf(Image image)  
    {  
      MemoryStream ms = new MemoryStream();  
      image.Save(ms, System.Drawing.Imaging.ImageFormat.Jpeg);  
 
      byte[] payload = ms.ToArray();  
 
      return payload;  
    }  
  }