上传android摄像头直播视频到RTP/RTSP服务器

Question

我已经做了适当的研究，但仍然缺乏关于我想要实现的东西的信息。

所以我想编程一个应用程序，在那里用户可以录制视频，并立即(现场)上传视频到RTP/RTSP服务器。服务器端不会成为问题。我不清楚的是如何在手机端实现这一点。

到目前为止，我的研究是，我必须将录制时的视频写入本地套接字，而不是文件，因为3gp文件如果写入文件，则在最终完成之前无法访问(当视频停止并且关于长度和其他内容的标题信息已写入视频时)。

当套接字接收到连续数据时，我需要将其包装在RTP包中，并将其发送到远程服务器。我可能还必须先做基本的编码(这还不是很重要)。

到目前为止，有没有人知道这个理论是否正确。我还想知道是否有人可以给我一些类似方法的代码片段，特别是在将视频动态发送到服务器时。我还不确定该怎么做。

非常感谢并致以最良好的问候

Answer 1

你的总体方法听起来是正确的，但有几件事你需要考虑。

所以我想编程一个应用程序，在那里用户可以录制视频，并立即(现场)上传视频到实时传输协议/实时传输协议服务器。

我假设你想上传到服务器，这样它就可以将内容重新分发给多个客户端？

• 你将如何处理到RTSP服务器的RTP会话的信令/建立？您需要以某种方式通知RTSP服务器，用户将上传实时媒体，以便它可以打开适当的RTP/RTCP套接字等。
• 您将如何处理身份验证？多个客户端设备-

到目前为止，我的研究是，我必须将录制时的视频写入本地套接字，而不是文件，因为3gp文件如果写入文件，则在最终完成之前无法访问(当视频停止并且关于长度和其他内容的头部信息已写入视频时)。

通过RTP/RTCP实时发送帧是正确的方法。当捕获设备捕获每一帧时，您需要对其进行编码/压缩并通过套接字发送。3gp和mp4一样，是一种用于文件存储的容器格式。对于实时捕获，不需要写入文件。唯一有意义的时候是，例如，在HTTP Live Streaming或DASH方法中，媒体在通过HTTP提供服务之前被写入传输流或mp4文件。

当套接字接收到连续的数据时，我需要将其封装在

包中，并将其发送到远程服务器。我可能还必须先做基本的编码(这还不是很重要)。

我不同意，编码是非常重要的，否则你可能永远不会设法发送视频，而且你将不得不处理诸如成本(通过移动网络)和取决于分辨率和帧率的媒体的绝对数量等问题。

，如果这个理论到目前为止是正确的，有人知道吗？我还想知道是否有人可以给我一些类似方法的代码片段，特别是在将视频动态发送到服务器时。我还不确定该怎么做。

以spydroid开源项目为起点。它包含许多必要的步骤，包括如何配置编码器、打包为RTP、发送RTCP以及一些RTSP服务器功能。Spydroid设置RTSP服务器，以便一旦使用RTSP客户端(如VLC )建立RTSP会话，就会对媒体进行编码和发送。由于您的应用程序是由想要将媒体发送到服务器的电话用户驱动的，因此您可能需要考虑另一种开始发送的方法，即使您向服务器发送某种类型的消息，例如像spydroid中那样建立RTSP会话。

Answer 2

一年前，我开发了一款android应用程序，可以使用tcp上的rtsp将其摄像头/麦克风传输到wowza媒体服务器。

一般的方法是创建unix套接字，获取其文件描述符，并将其提供给android媒体录制器组件。然后指示媒体记录器将mp4/h264格式的摄像机视频记录到该文件描述符中。现在，您的应用程序读取客户端套接字，解析mp4以移除报头并从中获取iframes，并将其动态包装到rtsp流中。

对声音(通常是AAC)也可以做一些类似的事情。当然，你必须自己处理时间戳，而整个方法中最棘手的事情是视频/音频同步。

所以这是它的第一部分。可以被称为rtspsocket的东西。它在connect方法中与媒体服务器进行协商，之后您可以将流本身写入其中。我稍后会展示它。

package com.example.android.streaming.streaming.rtsp;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.io.UnsupportedEncodingException;
import java.math.BigInteger;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.SocketException;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.Locale;
import java.util.concurrent.ConcurrentHashMap;

import android.util.Base64;
import android.util.Log;

import com.example.android.streaming.StreamingApp;
import com.example.android.streaming.streaming.Session;
import com.example.android.streaming.BuildConfig;

public class RtspSocket extends Socket {
    public static final int RTSP_HEADER_LENGTH = 4;
    public static final int RTP_HEADER_LENGTH = 12;
    public static final int MTU = 1400;

    public static final int PAYLOAD_OFFSET = RTSP_HEADER_LENGTH + RTP_HEADER_LENGTH;
    public static final int RTP_OFFSET = RTSP_HEADER_LENGTH;

    private ConcurrentHashMap<String, String> headerMap = new ConcurrentHashMap<String, String>();

    static private final String kCRLF = "\r\n";

    // RTSP request format strings
    static private final String kOptions = "OPTIONS %s RTSP/1.0\r\n";
    static private final String kDescribe = "DESCRIBE %s RTSP/1.0\r\n";
    static private final String kAnnounce = "ANNOUNCE %s RTSP/1.0\r\n";
    static private final String kSetupPublish = "SETUP %s/trackid=%d RTSP/1.0\r\n";
    @SuppressWarnings("unused")
    static private final String kSetupPlay = "SETUP %s/trackid=%d RTSP/1.0\r\n";
    static private final String kRecord = "RECORD %s RTSP/1.0\r\n";
    static private final String kPlay = "PLAY %s RTSP/1.0\r\n";
    static private final String kTeardown = "TEARDOWN %s RTSP/1.0\r\n";

    // RTSP header format strings
    static private final String kCseq = "Cseq: %d\r\n";
    static private final String kContentLength = "Content-Length: %d\r\n";
    static private final String kContentType = "Content-Type: %s\r\n";
    static private final String kTransport = "Transport: RTP/AVP/%s;unicast;mode=%s;%s\r\n";
    static private final String kSession = "Session: %s\r\n";
    static private final String kRange = "range: %s\r\n";
    static private final String kAccept = "Accept: %s\r\n";
    static private final String kAuthBasic = "Authorization: Basic %s\r\n";
    static private final String kAuthDigest = "Authorization: Digest username=\"%s\",realm=\"%s\",nonce=\"%s\",uri=\"%s\",response=\"%s\"\r\n";

    // RTSP header keys
    static private final String kSessionKey = "Session";
    static private final String kWWWAuthKey = "WWW-Authenticate";

    byte header[] = new byte[RTSP_MAX_HEADER + 1];
    static private final int RTSP_MAX_HEADER = 4095;
    static private final int RTSP_MAX_BODY = 4095;

    static private final int RTSP_RESP_ERR = -6;
    // static private final int RTSP_RESP_ERR_SESSION = -7;
    static public final int RTSP_OK = 200;
    static private final int RTSP_BAD_USER_PASS = 401;

    static private final int SOCK_ERR_READ = -5;

    /* Number of channels including control ones. */
    private int channelCount = 0;

    /* RTSP negotiation cmd seq counter */
    private int seq = 0;

    private String authentication = null;
    private String session = null;

    private String path = null;
    private String url = null;
    private String user = null;
    private String pass = null;
    private String sdp = null;

    private byte[] buffer = new byte[MTU];

    public RtspSocket() {
        super();
        try {
            setTcpNoDelay(true);
            setSoTimeout(60000);
        } catch (SocketException e) {
            Log.e(StreamingApp.TAG, "Failed to set socket params.");
        }
        buffer[RTSP_HEADER_LENGTH] = (byte) Integer.parseInt("10000000", 2);
    }

    public byte[] getBuffer() {
        return buffer;
    }

    public static final void setLong(byte[] buffer, long n, int begin, int end) {
        for (end--; end >= begin; end--) {
            buffer[end] = (byte) (n % 256);
            n >>= 8;
        }
    }

    public void setSequence(int seq) {
        setLong(buffer, seq, RTP_OFFSET + 2, RTP_OFFSET + 4);
    }

    public void setSSRC(int ssrc) {
        setLong(buffer, ssrc, RTP_OFFSET + 8, RTP_OFFSET + 12);
    }

    public void setPayload(int payload) {
        buffer[RTP_OFFSET + 1] = (byte) (payload & 0x7f);
    }

    public void setRtpTimestamp(long timestamp) {
        setLong(buffer, timestamp, RTP_OFFSET + 4, RTP_OFFSET + 8);
    }

    /** Sends the RTP packet over the network */
    private void send(int length, int stream) throws IOException {
        buffer[0] = '$';
        buffer[1] = (byte) stream;
        setLong(buffer, length, 2, 4);
        OutputStream s = getOutputStream();
        s.write(buffer, 0, length + RTSP_HEADER_LENGTH);
        s.flush();
    }

    public void sendReport(int length, int ssrc, int stream) throws IOException {
        setPayload(200);
        setLong(buffer, ssrc, RTP_OFFSET + 4, RTP_OFFSET + 8);
        send(length + RTP_HEADER_LENGTH, stream);
    }

    public void sendData(int length, int ssrc, int seq, int payload, int stream, boolean last) throws IOException {
        setSSRC(ssrc);
        setSequence(seq);
        setPayload(payload);
        buffer[RTP_OFFSET + 1] |= (((last ? 1 : 0) & 0x01) << 7);
        send(length + RTP_HEADER_LENGTH, stream);
    }

    public int getChannelCount() {
        return channelCount;
    }

    private void write(String request) throws IOException {
        try {
            String asci = new String(request.getBytes(), "US-ASCII");
            OutputStream out = getOutputStream();
            out.write(asci.getBytes());
        } catch (IOException e) {
            throw new IOException("Error writing to socket.");
        }
    }

    private String read() throws IOException {
        String response = null;
        try {
            InputStream in = getInputStream();
            int i = 0, len = 0, crlf_count = 0;
            boolean parsedHeader = false;

            for (; i < RTSP_MAX_BODY && !parsedHeader && len > -1; i++) {
                len = in.read(header, i, 1);
                if (header[i] == '\r' || header[i] == '\n') {
                    crlf_count++;
                    if (crlf_count == 4)
                        parsedHeader = true;
                } else {
                    crlf_count = 0;
                }
            }
            if (len != -1) {
                len = i;
                header[len] = '\0';
                response = new String(header, 0, len, "US-ASCII");
            }
        } catch (IOException e) {
            throw new IOException("Connection timed out. Check your network settings.");
        }
        return response;
    }

    private int parseResponse(String response) {
        String[] lines = response.split(kCRLF);
        String[] items = response.split(" ");
        String tempString, key, value;

        headerMap.clear();
        if (items.length < 2)
            return RTSP_RESP_ERR;
        int responseCode = RTSP_RESP_ERR;
        try {
            responseCode = Integer.parseInt(items[1]);
        } catch (Exception e) {
            Log.w(StreamingApp.TAG, e.getMessage());
            Log.w(StreamingApp.TAG, response);
        }
        if (responseCode == RTSP_RESP_ERR)
            return responseCode;

        // Parse response header into key value pairs.
        for (int i = 1; i < lines.length; i++) {
            tempString = lines[i];

            if (tempString.length() == 0)
                break;

            int idx = tempString.indexOf(":");

            if (idx == -1)
                continue;

            key = tempString.substring(0, idx);
            value = tempString.substring(idx + 1);
            headerMap.put(key, value);
        }

        tempString = headerMap.get(kSessionKey);
        if (tempString != null) {
            // Parse session
            items = tempString.split(";");
            tempString = items[0];
            session = tempString.trim();
        }

        return responseCode;
    }

    private void generateBasicAuth() throws UnsupportedEncodingException {
        String userpass = String.format("%s:%s", user, pass);
        authentication = String.format(kAuthBasic, Base64.encodeToString(userpass.getBytes("US-ASCII"), Base64.DEFAULT));
    }

    public static String md5(String s) {
        MessageDigest digest;
        try {
            digest = MessageDigest.getInstance("MD5");
            digest.update(s.getBytes(), 0, s.length());
            String hash = new BigInteger(1, digest.digest()).toString(16);
            return hash;
        } catch (NoSuchAlgorithmException e) {
            e.printStackTrace();
        }
        return "";
    }

    static private final int CC_MD5_DIGEST_LENGTH = 16;

    private String md5HexDigest(String input) {
        byte digest[] = md5(input).getBytes();
        String result = new String();
        for (int i = 0; i < CC_MD5_DIGEST_LENGTH; i++)
            result = result.concat(String.format("%02x", digest[i]));
        return result;
    }

    private void generateDigestAuth(String method) {
        String nonce, realm;
        String ha1, ha2, response;

        // WWW-Authenticate: Digest realm="Streaming Server",
        // nonce="206351b944cb28fe37a0794848c2e36f"
        String wwwauth = headerMap.get(kWWWAuthKey);
        int idx = wwwauth.indexOf("Digest");
        String authReq = wwwauth.substring(idx + "Digest".length() + 1);

        if (BuildConfig.DEBUG)
            Log.d(StreamingApp.TAG, String.format("Auth Req: %s", authReq));

        String[] split = authReq.split(",");
        realm = split[0];
        nonce = split[1];

        split = realm.split("=");
        realm = split[1];
        realm = realm.substring(1, 1 + realm.length() - 2);

        split = nonce.split("=");
        nonce = split[1];
        nonce = nonce.substring(1, 1 + nonce.length() - 2);

        if (BuildConfig.DEBUG) {
            Log.d(StreamingApp.TAG, String.format("realm=%s", realm));
            Log.d(StreamingApp.TAG, String.format("nonce=%s", nonce));
        }

        ha1 = md5HexDigest(String.format("%s:%s:%s", user, realm, pass));
        ha2 = md5HexDigest(String.format("%s:%s", method, url));
        response = md5HexDigest(String.format("%s:%s:%s", ha1, nonce, ha2));
        authentication = md5HexDigest(String.format(kAuthDigest, user, realm, nonce, url, response));
    }

    private int options() throws IOException {
        seq++;
        StringBuilder request = new StringBuilder();
        request.append(String.format(kOptions, url));
        request.append(String.format(kCseq, seq));
        request.append(kCRLF);
        if (BuildConfig.DEBUG)
            Log.d(StreamingApp.TAG, "--- OPTIONS Request ---\n\n" + request);
        write(request.toString());
        String response = read();
        if (response == null)
            return SOCK_ERR_READ;
        if (BuildConfig.DEBUG)
            Log.d(StreamingApp.TAG, "--- OPTIONS Response ---\n\n" + response);
        return parseResponse(response);
    }

    @SuppressWarnings("unused")
    private int describe() throws IOException {
        seq++;
        StringBuilder request = new StringBuilder();
        request.append(String.format(kDescribe, url));
        request.append(String.format(kAccept, "application/sdp"));
        request.append(String.format(kCseq, seq));
        request.append(kCRLF);
        if (BuildConfig.DEBUG)
            Log.d(StreamingApp.TAG, "--- DESCRIBE Request ---\n\n" + request);
        write(request.toString());
        String response = read();
        if (response == null)
            return SOCK_ERR_READ;
        if (BuildConfig.DEBUG)
            Log.d(StreamingApp.TAG, "--- DESCRIBE Response ---\n\n" + response);
        return parseResponse(response);
    }

    private int recurseDepth = 0;

    private int announce() throws IOException {
        seq++;
        recurseDepth = 0;
        StringBuilder request = new StringBuilder();
        request.append(String.format(kAnnounce, url));
        request.append(String.format(kCseq, seq));
        request.append(String.format(kContentLength, sdp.length()));
        request.append(String.format(kContentType, "application/sdp"));
        request.append(kCRLF);
        if (sdp.length() > 0)
            request.append(sdp);
        if (BuildConfig.DEBUG)
            Log.d(StreamingApp.TAG, "--- ANNOUNCE Request ---\n\n" + request);
        write(request.toString());
        String response = read();
        if (response == null)
            return SOCK_ERR_READ;
        if (BuildConfig.DEBUG)
            Log.d(StreamingApp.TAG, "--- ANNOUNCE Response ---\n\n" + response);

        int ret = parseResponse(response);
        if (ret == RTSP_BAD_USER_PASS && recurseDepth == 0) {
            String wwwauth = headerMap.get(kWWWAuthKey);
            if (wwwauth != null) {
                if (BuildConfig.DEBUG)
                    Log.d(StreamingApp.TAG, String.format("WWW Auth Value: %s", wwwauth));
                int idx = wwwauth.indexOf("Basic");
                recurseDepth++;

                if (idx != -1) {
                    generateBasicAuth();
                } else {
                    // We are assuming Digest here.
                    generateDigestAuth("ANNOUNCE");
                }

                ret = announce();
                recurseDepth--;
            }
        }
        return ret;
    }

    private int setup(int trackId) throws IOException {
        seq++;
        recurseDepth = 0;
        StringBuilder request = new StringBuilder();
        request.append(String.format(kSetupPublish, url, trackId));
        request.append(String.format(kCseq, seq));

        /* One channel for rtp (data) and one for rtcp (control) */
        String tempString = String.format(Locale.getDefault(), "interleaved=%d-%d", channelCount++, channelCount++);

        request.append(String.format(kTransport, "TCP", "record", tempString));
        request.append(kCRLF);
        if (BuildConfig.DEBUG)
            Log.d(StreamingApp.TAG, "--- SETUP Request ---\n\n" + request);
        write(request.toString());
        String response = read();
        if (response == null)
            return SOCK_ERR_READ;
        if (BuildConfig.DEBUG)
            Log.d(StreamingApp.TAG, "--- SETUP Response ---\n\n" + response);

        int ret = parseResponse(response);
        if (ret == RTSP_BAD_USER_PASS && recurseDepth == 0) {
            String wwwauth = headerMap.get(kWWWAuthKey);
            if (wwwauth != null) {
                if (BuildConfig.DEBUG)
                    Log.d(StreamingApp.TAG, String.format("WWW Auth Value: %s", wwwauth));
                int idx = wwwauth.indexOf("Basic");
                recurseDepth++;

                if (idx != -1) {
                    generateBasicAuth();
                } else {
                    // We are assuming Digest here.
                    generateDigestAuth("SETUP");
                }

                ret = setup(trackId);
                authentication = null;
                recurseDepth--;
            }
        }
        return ret;
    }

    private int record() throws IOException {
        seq++;
        recurseDepth = 0;
        StringBuilder request = new StringBuilder();
        request.append(String.format(kRecord, url));
        request.append(String.format(kCseq, seq));
        request.append(String.format(kRange, "npt=0.000-"));
        if (authentication != null)
            request.append(authentication);
        if (session != null)
            request.append(String.format(kSession, session));
        request.append(kCRLF);
        if (BuildConfig.DEBUG)
            Log.d(StreamingApp.TAG, "--- RECORD Request ---\n\n" + request);
        write(request.toString());
        String response = read();
        if (response == null)
            return SOCK_ERR_READ;
        if (BuildConfig.DEBUG)
            Log.d(StreamingApp.TAG, "--- RECORD Response ---\n\n" + response);
        int ret = parseResponse(response);
        if (ret == RTSP_BAD_USER_PASS && recurseDepth == 0) {
            String wwwauth = headerMap.get(kWWWAuthKey);
            if (wwwauth != null) {
                if (BuildConfig.DEBUG)
                    Log.d(StreamingApp.TAG, String.format("WWW Auth Value: %s", wwwauth));
                int idx = wwwauth.indexOf("Basic");
                recurseDepth++;

                if (idx != -1) {
                    generateBasicAuth();
                } else {
                    // We are assuming Digest here.
                    generateDigestAuth("RECORD");
                }

                ret = record();
                authentication = null;
                recurseDepth--;
            }
        }
        return ret;
    }

    @SuppressWarnings("unused")
    private int play() throws IOException {
        seq++;
        recurseDepth = 0;
        StringBuilder request = new StringBuilder();
        request.append(String.format(kPlay, url));
        request.append(String.format(kCseq, seq));
        request.append(String.format(kRange, "npt=0.000-"));
        if (authentication != null)
            request.append(authentication);
        if (session != null)
            request.append(String.format(kSession, session));
        request.append(kCRLF);
        if (BuildConfig.DEBUG)
            Log.d(StreamingApp.TAG, "--- PLAY Request ---\n\n" + request);
        write(request.toString());
        String response = read();
        if (response == null)
            return SOCK_ERR_READ;
        if (BuildConfig.DEBUG)
            Log.d(StreamingApp.TAG, "--- PLAY Response ---\n\n" + response);
        int ret = parseResponse(response);
        if (ret == RTSP_BAD_USER_PASS && recurseDepth == 0) {
            String wwwauth = headerMap.get(kWWWAuthKey);
            if (wwwauth != null) {
                if (BuildConfig.DEBUG)
                    Log.d(StreamingApp.TAG, String.format("WWW Auth Value: %s", wwwauth));
                int idx = wwwauth.indexOf("Basic");
                recurseDepth++;

                if (idx != -1) {
                    generateBasicAuth();
                } else {
                    // We are assuming Digest here.
                    generateDigestAuth("PLAY");
                }

                ret = record();
                authentication = null;
                recurseDepth--;
            }
        }
        return ret;
    }

    private int teardown() throws IOException {
        seq++;
        recurseDepth = 0;
        StringBuilder request = new StringBuilder();
        request.append(String.format(kTeardown, url));
        request.append(String.format(kCseq, seq));
        if (authentication != null)
            request.append(authentication);
        if (session != null)
            request.append(String.format(kSession, session));
        request.append(kCRLF);
        if (BuildConfig.DEBUG)
            Log.d(StreamingApp.TAG, "--- TEARDOWN Request ---\n\n" + request);
        write(request.toString());
        String response = read();
        if (response == null)
            return SOCK_ERR_READ;
        if (BuildConfig.DEBUG)
            Log.d(StreamingApp.TAG, "--- TEARDOWN Response ---\n\n" + response);
        int ret = parseResponse(response);
        if (ret == RTSP_BAD_USER_PASS && recurseDepth == 0) {
            String wwwauth = headerMap.get(kWWWAuthKey);
            if (wwwauth != null) {
                if (BuildConfig.DEBUG)
                    Log.d(StreamingApp.TAG, String.format("WWW Auth Value: %s", wwwauth));
                int idx = wwwauth.indexOf("Basic");
                recurseDepth++;

                if (idx != -1) {
                    generateBasicAuth();
                } else {
                    // We are assuming Digest here.
                    generateDigestAuth("TEARDOWN");
                }

                ret = record();
                authentication = null;
                recurseDepth--;
            }
        }
        return ret;
    }

    public void connect(String dest, int port, Session session) throws IOException {
        int trackId = 1;
        int responseCode;

        if (isConnected())
            return;

        if (!session.hasAudioTrack() && !session.hasVideoTrack())
            throw new IOException("No tracks found in session.");

        InetSocketAddress addr = null;
        try {
            addr = new InetSocketAddress(dest, port);
        } catch (Exception e) {
            throw new IOException("Failed to resolve rtsp server address.");
        }

        this.sdp = session.getSDP();
        this.user = session.getUser();
        this.pass = session.getPass();
        this.path = session.getPath();
        this.url = String.format("rtsp://%s:%d%s", dest, addr.getPort(), this.path);

        try {
            super.connect(addr);
        } catch (IOException e) {
            throw new IOException("Failed to connect rtsp server.");
        }

        responseCode = announce();
        if (responseCode != RTSP_OK) {
            close();
            throw new IOException("RTSP announce failed: " + responseCode);
        }

        responseCode = options();
        if (responseCode != RTSP_OK) {
            close();
            throw new IOException("RTSP options failed: " + responseCode);
        }

        /* Setup audio */
        if (session.hasAudioTrack()) {
            session.getAudioTrack().setStreamId(channelCount);
            responseCode = setup(trackId++);
            if (responseCode != RTSP_OK) {
                close();
                throw new IOException("RTSP video failed: " + responseCode);
            }
        }

        /* Setup video */
        if (session.hasVideoTrack()) {
            session.getVideoTrack().setStreamId(channelCount);
            responseCode = setup(trackId++);
            if (responseCode != RTSP_OK) {
                close();
                throw new IOException("RTSP audio setup failed: " + responseCode);
            }
        }

        responseCode = record();
        if (responseCode != RTSP_OK) {
            close();
            throw new IOException("RTSP record failed: " + responseCode);
        }
    }

    public void close() throws IOException {
        if (!isConnected())
            return;
        teardown();
        super.close();
    }
}

Answer 3

我试图达到同样的效果(但由于缺乏经验而放弃了)。我的方法是使用ffmpeg和/或avlib，因为它已经有了可以工作的rtmp堆栈。所以从理论上讲，你需要做的就是将视频流路由到ffmpeg进程，再由ffmpeg进程发送到服务器。