red5源碼分析—客戶端連接

本博文開始分析red5服務器以及客戶端的源碼，選取的版本為最新的1.0.7。
red5發展到現在，可以兼容很多流媒體傳輸協議，例如rtmp、rtmpt等等。本博文只分析rtmp協議，其他協議看看以后有沒有時間研究吧。
red5的服務器啟動有好幾種方式，standalone、tomcat、jetty等等。本博文只分析standalone的啟動方式。本博文假設客戶端為red5 rtmpclient。
red5 client和server的下載地址如下
https://github.com/Red5
首先看一段網上很常見的red5的客戶端代碼，如下所示

public class RtmpClientTest extends RTMPClient implements INetStreamEventHandler, IPendingServiceCallback, IEventDispatcher { private ConcurrentLinkedQueue<IMessage> frameBuffer = new ConcurrentLinkedQueue<IMessage>();String host = "127.0.0.1";String app = "red5test"; int port = 1935; public RtmpClientTest() { super(); Map<String, Object> map = makeDefaultConnectionParams(host, 1935, "red5test"); connect(host, 1935, map, this); } @Override public void dispatchEvent(IEvent arg0) { } @Override public void resultReceived(IPendingServiceCall call) { Object result = call.getResult(); if (result instanceof ObjectMap) { if ("connect".equals(call.getServiceMethodName())) { createStream(this); } } else { if ("createStream".equals(call.getServiceMethodName())) { if (result instanceof Integer) { Integer streamIdInt = (Integer) result; // int streamId = streamIdInt.intValue(); // publish(streamId, "testgio2", "live", this); invoke("getRoomsInfo", this); } else { disconnect(); } } else if ("getRoomsInfo".equals(call.getServiceMethodName())) { ArrayList<String> list = (ArrayList<String>) result; for (int i = 0; i < list.size(); i++) { System.out.println(list.get(i)); } } } } @Override public void onStreamEvent(Notify arg0) { ObjectMap<?, ?> map = (ObjectMap<?, ?>) notify.getCall().getArguments()[0];String code = (String) map.get("code");if (StatusCodes.NS_PUBLISH_START.equals(code)) {IMessage message = null;while ((message = frameBuffer.poll()) != null) {this.publishStreamData(streamId, message);}} else if (StatusCodes.NS_UNPUBLISHED_SUCCESS.equals(code)) {}} @Override public void connectionOpened(RTMPConnection conn, RTMP state) { super.connectionOpened(conn, state); } public static void main(String[] args) { new RtmpClientTest(); } }

RtmpClientTest實現的三個接口INetStreamEventHandler、IPendingServiceCallback、IEventDispatcher和回調函數有關，后面的章節會分析到。
首先來看RtmpClientTest的構造函數，其父類RTMPClient的構造函數如下

public RTMPClient() {ioHandler = new RTMPMinaIoHandler();ioHandler.setHandler(this);}

red5客戶端使用mina框架來封裝Java Nio，關于mina框架的源碼分析請查看博主的mina源碼分析系列文章。這里有兩個handler，RTMPMinaIoHandler和mina框架有關，另一個handler就是RTMPClient自身，因為其繼承自BaseRTMPClientHandler，BaseRTMPClientHandler和業務有關。
回到RtmpClientTest構造函數，接下來調用connect進行連接，connect函數實現在RTMPClient的父類BaseRTMPClientHandler中，

public void connect(String server, int port, Map<String, Object> connectionParams, IPendingServiceCallback connectCallback) {connect(server, port, connectionParams, connectCallback, null);}public void connect(String server, int port, Map<String, Object> connectionParams, IPendingServiceCallback connectCallback, Object[] connectCallArguments) {this.connectionParams = connectionParams;this.connectArguments = connectCallArguments;if (!connectionParams.containsKey("objectEncoding")) {connectionParams.put("objectEncoding", 0);}this.connectCallback = connectCallback;startConnector(server, port);}

connect函數一開始作了一些簡單的設置，最后通過startConnector與服務器建立連接。startConnector定義在RTMPClient中，

protected void startConnector(String server, int port) {socketConnector = new NioSocketConnector();socketConnector.setHandler(ioHandler);future = socketConnector.connect(new InetSocketAddress(server, port));future.addListener(new IoFutureListener<ConnectFuture>() {public void operationComplete(ConnectFuture future) {try {session = future.getSession();} catch (Throwable e) {socketConnector.dispose(false);handleException(e);}}});future.awaitUninterruptibly(CONNECTOR_WORKER_TIMEOUT);}

這里主要構造了一個NioSocketConnector，并調用其connect函數。connect函數會使用mina框架與服務器建立連接，下一章會分析服務器如何處理客戶端的連接請求。當與服務器建立完連接（TCP連接）時，根據mina框架的源碼，會回調mina框架中IoHandler的處理函數，也即前面注冊的RTMPMinaIoHandler的sessionCreated和sessionOpened函數，下面依次來看。

一. sessionCreated

sessionCreated的代碼如下，

public void sessionCreated(IoSession session) throws Exception {session.getFilterChain().addFirst("rtmpeFilter", new RTMPEIoFilter());RTMPMinaConnection conn = createRTMPMinaConnection();conn.setIoSession(session);session.setAttribute(RTMPConnection.RTMP_SESSION_ID, conn.getSessionId());OutboundHandshake outgoingHandshake = new OutboundHandshake();session.setAttribute(RTMPConnection.RTMP_HANDSHAKE, outgoingHandshake);if (enableSwfVerification) {String swfUrl = (String) handler.getConnectionParams().get("swfUrl");if (!StringUtils.isEmpty(swfUrl)) {outgoingHandshake.initSwfVerification(swfUrl);}}session.setAttribute(RTMPConnection.RTMP_HANDLER, handler);handler.setConnection((RTMPConnection) conn);}

sessionCreated函數首先向mina框架中添加一個過濾器RTMPEIoFilter，該過濾器用來處理RTMP協議的握手過程，具體的RTMP協議可以從網上下載。sessionCreated接著創建一個RTMPMinaConnection并進行相應的設置，

protected RTMPMinaConnection createRTMPMinaConnection() {return (RTMPMinaConnection) RTMPConnManager.getInstance().createConnection(RTMPMinaConnection.class);}

RTMPConnManager使用單例模式，其createConnection函數如下，

public RTMPConnection createConnection(Class<?> connCls) {RTMPConnection conn = null;if (RTMPConnection.class.isAssignableFrom(connCls)) {try {conn = createConnectionInstance(connCls);connMap.put(conn.getSessionId(), conn);} catch (Exception ex) {}}return conn;}public RTMPConnection createConnectionInstance(Class<?> cls) throws Exception {RTMPConnection conn = null;if (cls == RTMPMinaConnection.class) {conn = (RTMPMinaConnection) cls.newInstance();} else if (cls == RTMPTClientConnection.class) {conn = (RTMPTClientConnection) cls.newInstance();} else {conn = (RTMPConnection) cls.newInstance();}conn.setMaxHandshakeTimeout(maxHandshakeTimeout);conn.setMaxInactivity(maxInactivity);conn.setPingInterval(pingInterval);ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();executor.setCorePoolSize(1);executor.setDaemon(true);executor.setMaxPoolSize(1);executor.setQueueCapacity(executorQueueCapacity);executor.initialize();conn.setExecutor(executor);return conn;}

這里就是實例化一個RTMPMinaConnection，創建ThreadPoolTaskExecutor并進行相應的設置，最后添加進connMap中。注意每個RTMPMinaConnection的SessionId是隨機生成的。
回到sessionCreated中，接下來設置剛剛構造的RTMPMinaConnection，以及其SessionId，然后創建OutboundHandshake用于RTMP協議的握手，握手結束后該OutboundHandshake將會從session中移除，最后設置handler和RTMPMinaConnection。

二. sessionOpened

再來看RTMPMinaIoHandler的sessionOpened函數，代碼如下

public void sessionOpened(IoSession session) throws Exception {super.sessionOpened(session);RTMPHandshake handshake = (RTMPHandshake) session.getAttribute(RTMPConnection.RTMP_HANDSHAKE);IoBuffer clientRequest1 = ((OutboundHandshake) handshake).generateClientRequest1();session.write(clientRequest1);}

這里根據從session中獲得剛剛在sessionCreated中創建的OutboundHandshake，調用其generateClientRequest1函數生成第一次握手請求的數據，通過write函數發送給服務器。generateClientRequest1函數和具體的協議相關，這里就不繼續往下看了。

總結一下，本章分析了如何創建一個RTMPClient并建立與服務器的TCP連接，然后發送第一次握手請求開始與服務器建立RTMP連接，下一章開始分析red5服務器端對應的連接函數。

總結

以上是生活随笔為你收集整理的red5源码分析---1的全部內容，希望文章能夠幫你解決所遇到的問題。

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