Zookeeper 客户端源码吐血总结

目录
一、几个重要的类
二、JAVA的基础知识
三、大致了解
四从入门到放弃的讲解
Code1：ZK
Code2：创建 Zookeeper实例,实例化ClientCnxn,实例化ClientCnxnSocketNIO
Code3：实例化ClientCnxnSocketNIO (which extends ClientCnxnSocket)
Code4：ClientCnxn的具体实例化
Code5：SendThread的具体实例化
Code6：EventThread的具体实例化
Code 7：SendThread核心run流程
Code 8：startConnect()
Code 9： clientCnxnSocket.connect
Code10：registerAndConnect()
Code11：primeConnection()
Code 12：doTransport()
Code13：findSendablePacket()
Code14：IO write
Code15：createBB()
Code 16：IO read
Code 17： readResponse
Code 18：EventThread run:

源码: Zookeeper 3.4.6.jar(吐血总结)

一、几个重要的类

    1） ZookeeperMain: main函数为入口，由zkCli.sh脚本调用启动2)  ZooKeeper：客户端入口3)  ZooKeeper.SendThread: IO线程4） ZooKeeper.EventThread: 事件处理线程，处理各类消息callback5） ClientCnxn: 客户端与服务器端交互的主要类6） ClientCnxnSocketNIO：继承自ClientCnxnSocket，专门处理IO, 利用JAVANIO7） Watcher: 用于监控Znode节点9） WatcherManager：用来管理Watcher，管理了ZK Client绑定的所有Watcher。

二、JAVA的基础知识

1）JAVA多线程
2）JAVANIO:  可参考：http://blog.csdn.net/cnh294141800/article/details/52996819
3）Socket编程（稍微了解即可）
4）JLine: 是一个用来处理控制台输入的Java类库

三、大致了解

上图就是对Zookeeper源码一个最好的解释，
（1） Client端发送Request(封装成Packet)请求到Zookeeper
（2） Zookeeper处理Request并将该请求放入Outgoing Queue(顾名思义，外出队列，就是让Zookeeper服务器处理的队列)，
（3） Zookeeper端处理Outgoing Queue，并将该事件移到Pending Queue中
（4） Zookeeper端消费Pending Queue，并调用finishPacket(),生成Event
（5） EventThread线程消费Event事件,并且处理Watcher.

四从入门到放弃的讲解
（1）应用提供watch实例(new MyWatcher(null))

private class MyWatcher implements Watcher {  // 默认Watcher public void process(WatchedEvent event) {if (getPrintWatches()) {ZooKeeperMain.printMessage("WATCHER::");ZooKeeperMain.printMessage(event.toString());}}}

（2）实例化zookeeper
 实例化socket,默认使用ClientCnxnSocketNIO
 实例化ClientCnxn
 实例化SendThread
 实例化EventThread
Code1：ZK

zk = new ZooKeeper(host Integer.parseInt(cl.getOption("timeout")),new MyWatcher(), readOnly);   //  初始化ZK

Code2：

public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher,boolean canBeReadOnly)throws IOException{…watchManager.defaultWatcher = watcher;    // 设置defaultWatcher 为 MyWatcherConnectStringParser connectStringParser = new ConnectStringParser(connectString);   // 解析-server 获取 IP以及PORTHostProvider hostProvider = new StaticHostProvider(connectStringParser.getServerAddresses());cnxn = new ClientCnxn(connectStringParser.getChrootPath(),hostProvider, sessionTimeout, this, watchManager,getClientCnxnSocket(), canBeReadOnly);  // 创建 ClientCnxn实例cnxn.start();   // 启动cnxn中的SendThread and EventThread进程}

Code3：实例化ClientCnxnSocketNIO (which extends ClientCnxnSocket)

private static ClientCnxnSocket getClientCnxnSocket() throws IOException {String clientCnxnSocketName = System.getProperty(ZOOKEEPER_CLIENT_CNXN_SOCKET);if (clientCnxnSocketName == null) {clientCnxnSocketName = ClientCnxnSocketNIO.class.getName();}try {return (ClientCnxnSocket) Class.forName(clientCnxnSocketName).newInstance();} catch (Exception e) {IOException ioe = new IOException("Couldn't instantiate "+ clientCnxnSocketName);ioe.initCause(e);throw ioe;}}

Code4：ClientCnxn的具体实例化
/* 另一个ClientCnxn构造函数，可见时sessionId=0

public ClientCnxn(String chrootPath, HostProvider hostProvider, int sessionTimeout, ZooKeeper zooKeeper,ClientWatchManager watcher, ClientCnxnSocket clientCnxnSocket, boolean canBeReadOnly)throws IOException {this(chrootPath, hostProvider, sessionTimeout, zooKeeper, watcher,clientCnxnSocket, 0, new byte[16], canBeReadOnly);
}
*/
public ClientCnxn(String chrootPath, HostProvider hostProvider, int sessionTimeout, ZooKeeper zooKeeper,ClientWatchManager watcher, ClientCnxnSocket clientCnxnSocket,long sessionId, byte[] sessionPasswd, boolean canBeReadOnly) {this.zooKeeper = zooKeeper;   this.watcher = watcher;this.sessionId = sessionId;this.sessionPasswd = sessionPasswd;this.sessionTimeout = sessionTimeout;//主机列表this.hostProvider = hostProvider;this.chrootPath = chrootPath;//连接超时connectTimeout = sessionTimeout / hostProvider.size();//读超时 readTimeout = sessionTimeout * 2 / 3;readOnly = canBeReadOnly;//初始化client2个核心线程，SendThread是client的IO核心线程，EventThread从SendThread里拿到eventsendThread = new SendThread(clientCnxnSocket);eventThread = new EventThread();}

Code5：SendThread的具体实例化

SendThread(ClientCnxnSocket clientCnxnSocket) {super(makeThreadName("-SendThread()"));state = States.CONNECTING;    // 将状态设置为连接状态（此时还未连接）this.clientCnxnSocket = clientCnxnSocket;setUncaughtExceptionHandler(uncaughtExceptionHandler);setDaemon(true); //设为守护线程}

Code6：EventThread的具体实例化

EventThread() {super(makeThreadName("-EventThread"));setUncaughtExceptionHandler(uncaughtExceptionHandler);setDaemon(true);}

至此所有对象实例化完成，然后启动SendThread、EventThread进程
（3）启动zookeeper
 启动SendThread
 连接服务器
 产生真正的socket，见ClientCnxnSocketNIO.createSock
 向select注册一个OP_CONNECT事件并连接服务器，由于是非阻塞连接，此时有可能并不会立即连上，如果连上就会调用SendThread.primeConnection初始化连接来注册读写事件，否则会在接下来的轮询select获取连接事件中处理
 复位socket的incomingBuffer
 连接成功后会产生一个connect型的请求发给服务，用于获取本次连接的sessionid
 进入循环等待来自应用的请求，如果没有就根据时间来ping 服务器
 启动EventThread
开始进入无限循环，从队列waitingEvents中获取事件，如果没有就阻塞等待

Code 7：SendThread核心run流程
可以对run进行抽象看待，流程如下

loop:- try:
     - - !isConnected()
     - - - connect()
     - - doTransport()
     - catch:
     - - cleanup()
 close()

先判断是否连接，没有连接则调用connect方法进行连接，有连接则直接使用；然后调用doTransport方法进行通信，若连接过程中出现异常，则调用cleanup()方法；最后关闭连接。

public void run() {while (state.isAlive()) {     // this != CLOSED && this != AUTH_FAILED; 刚才设置了首次状态为连接状态try {//如果还没连上，则启动连接程序if (!clientCnxnSocket.isConnected()) {   //所有的clientCnxnSocket都是clientCnxnSocketDIO实例//不是首次连接则休息1Sif(!isFirstConnect){  try {Thread.sleep(r.nextInt(1000));} catch (InterruptedException e) {LOG.warn("Unexpected exception", e);}}// don't re-establish connection if we are closingif (closing || !state.isAlive()) {break;}startConnect();// 启动连接clientCnxnSocket.updateLastSendAndHeard(); //更新Socket最后一次发送以及听到消息的时间}if (state.isConnected()) {// determine whether we need to send an AuthFailed event.if (zooKeeperSaslClient != null) {......  }// 下一次超时时间to = readTimeout - clientCnxnSocket.getIdleRecv();} else {// 如果还没连接上 重置当前剩余可连接时间to = connectTimeout - clientCnxnSocket.getIdleRecv();}// 连接超时if (to <= 0) {}// 判断是否 需要发送Ping心跳包if (state.isConnected()) {sendPing();}// If we are in read-only mode, seek for read/write serverif (state == States.CONNECTEDREADONLY) {}// The most important step. Do real IOclientCnxnSocket.doTransport(to, pendingQueue, outgoingQueue, ClientCnxn.this);} catch (Throwable e) {}}cleanup();...}
}

Code 8：startConnect()
// 具体实际连接部分

private void startConnect() throws IOException {state = States.CONNECTING;  //state 状态设置为连接InetSocketAddress addr; if (rwServerAddress != null) {addr = rwServerAddress;rwServerAddress = null;} else {addr = hostProvider.next(1000);}setName(getName().replaceAll("\\(.*\\)","(" + addr.getHostName() + ":" + addr.getPort() + ")"));if (ZooKeeperSaslClient.isEnabled()) {}logStartConnect(addr);   //写连接日志clientCnxnSocket.connect(addr);   //连接Socket}

Code 9： clientCnxnSocket.connect

void connect(InetSocketAddress addr) throws IOException {SocketChannel sock = createSock();  // 创建一个非阻塞空SocketChanneltry {registerAndConnect(sock, addr);  //注册并且连接sock到辣个addr } catch (IOException e) {….}initialized = false;/* Reset incomingBuffer*/lenBuffer.clear();incomingBuffer = lenBuffer;}
}

Code10：registerAndConnect()

void registerAndConnect(SocketChannel sock, InetSocketAddress addr) throws IOException {sockKey = sock.register(selector, SelectionKey.OP_CONNECT); //将socket注册到selector中boolean immediateConnect = sock.connect(addr);  //socket连接服务器if (immediateConnect) {sendThread.primeConnection();  //初始化连接事件}
}

Code11：primeConnection()

void primeConnection()  IOException {LOG.info("Socket connection established to "+ clientCnxnSocket.getRemoteSocketAddress()+ ", initiating session");isFirstConnect = false;  // 设置为非首次连接long sessId = (seenRwServerBefore) ? sessionId : 0; // 客户端默认sessionid为0// 创建连接request lastZxid 代表最新一次的节点ZXIDConnectRequest conReq = new ConnectRequest(0, lastZxid,sessionTimeout, sessId, sessionPasswd); // 线程安全占用outgoing synchronized (outgoingQueue) {…//组合成通讯层的Packet对象，添加到发送队列，对于ConnectRequest其requestHeader为null  outgoingQueue.addFirst(new Packet(null, null, conReq,null, null, readOnly));}//确保读写事件都监听，也就是设置成可读可写clientCnxnSocket.enableReadWriteOnly();if (LOG.isDebugEnabled()) {LOG.debug("Session establishment request sent on "+ clientCnxnSocket.getRemoteSocketAddress());}}

至此Channelsocket已经成功连接，并且已将连接请求做为队列放到Outgoing中。此时，需要再回头看Code7, 也就是一直在循环的SendThread部分。可以看到连接部分成功完成，接下来需要做doTransport()。// CnxnClientSocketNio

Code 12：doTransport()

void doTransport(int waitTimeOut, List<Packet> pendingQueue, LinkedList<Packet> outgoingQueue,ClientCnxn cnxn)throws IOException, InterruptedException {//selectselector.select(waitTimeOut);Set<SelectionKey> selected;synchronized (this) {selected = selector.selectedKeys();}// Everything below and until we get back to the select is// non blocking, so time is effectively a constant. That is// Why we just have to do this once, hereupdateNow();for (SelectionKey k : selected) {SocketChannel sc = ((SocketChannel) k.channel());//如果之前连接没有立马连上，则在这里处理OP_CONNECT事件if ((k.readyOps() & SelectionKey.OP_CONNECT) != 0) {if (sc.finishConnect()) {updateLastSendAndHeard();sendThread.primeConnection();}} //如果读写就位，则处理之else if ((k.readyOps() & (SelectionKey.OP_READ | SelectionKey.OP_WRITE)) != 0) {doIO(pendingQueue, outgoingQueue, cnxn);}}if (sendThread.getZkState().isConnected()) {synchronized(outgoingQueue) {//找到连接Packet并且将他放到队列头if (findSendablePacket(outgoingQueue,cnxn.sendThread.clientTunneledAuthenticationInProgress()) != null) {//  将要Channecl设置为可读enableWrite();}}}selected.clear();
}

Code13：findSendablePacket()

private Packet findSendablePacket(LinkedList<Packet> outgoingQueue,boolean clientTunneledAuthenticationInProgress) {synchronized (outgoingQueue) {..// Since client's authentication with server is in progress,// send only the null-header packet queued by primeConnection().// This packet must be sent so that the SASL authentication process// can proceed, but all other packets should wait until// SASL authentication completes.//因为Conn Packet需要发送到SASL authentication进行处理，其他Packet都需要等待直到该处理完成，//Conn Packet必须第一个处理，所以找出它并且把它放到OutgoingQueue头,也就是requestheader=null的辣个ListIterator<Packet> iter = outgoingQueue.listIterator();while (iter.hasNext()) {Packet p = iter.next();if (p.requestHeader == null) {// We've found the priming-packet. Move it to the beginning of the queue.iter.remove();  outgoingQueue.add(0, p); // 将连接放到outgogingQueue第一个return p;} else {// Non-priming packet: defer it until later, leaving it in the queue// until authentication completes.if (LOG.isDebugEnabled()) {LOG.debug("deferring non-priming packet: " + p +"until SASL authentication completes.");}}}// no sendable packet found.return null;}
}

然后就是最重要的IO部分：
 需要处理两类网络事件（读、写）

Code14：IO write

if (sockKey.isWritable()) {synchronized(outgoingQueue) {// 获得packetPacket p = findSendablePacket(outgoingQueue,cnxn.sendThread.clientTunneledAuthenticationInProgress());if (p != null) {updateLastSend();// If we already started writing p, p.bb will already existif (p.bb == null) {if ((p.requestHeader != null) &&(p.requestHeader.getType() != OpCode.ping) &&(p.requestHeader.getType() != OpCode.auth)) {//如果不是 连接事件，不是ping 事件，不是 认证时间 p.requestHeader.setXid(cnxn.getXid());}// 序列化p.createBB();}//将数据写入Channelsock.write(p.bb);// p.bb中如果没有内容 则表示发送成功if (!p.bb.hasRemaining()) {//发送数+1sentCount++;//将该P从队列中移除outgoingQueue.removeFirstOccurrence(p);//如果该事件不是连接事件，不是ping事件，不是认证事件， 则将他加入pending队列中if (p.requestHeader != null&& p.requestHeader.getType() != OpCode.ping&& p.requestHeader.getType() != OpCode.auth) {synchronized (pendingQueue) {pendingQueue.add(p);}}}}if (outgoingQueue.isEmpty()) {// No more packets to send: turn off write interest flag.// Will be turned on later by a later call to enableWrite(),// from within ZooKeeperSaslClient (if client is configured// to attempt SASL authentication), or in either doIO() or// in doTransport() if not.disableWrite();} else if (!initialized && p != null && !p.bb.hasRemaining()) {// On initial connection, write the complete connect request// packet, but then disable further writes until after// receiving a successful connection response.  If the// session is expired, then the server sends the expiration// response and immediately closes its end of the socket.  If// the client is simultaneously writing on its end, then the// TCP stack may choose to abort with RST, in which case the// client would never receive the session expired event.  See// http://docs.oracle.com/javase/6/docs/technotes/guides/net/articles/connection_release.htmldisableWrite();} else {// Just in caseenableWrite();}}}

Code15：createBB()

public void createBB() {try {ByteArrayOutputStream baos = new ByteArrayOutputStream();BinaryOutputArchive boa = BinaryOutputArchive.getArchive(baos);boa.writeInt(-1, "len"); // We'll fill this in later// 如果不是连接事件则设置协议头if (requestHeader != null) {requestHeader.serialize(boa, "header");}//设置协议体if (request instanceof ConnectRequest) {request.serialize(boa, "connect");// append "am-I-allowed-to-be-readonly" flagboa.writeBool(readOnly, "readOnly");} else if (request != null) {request.serialize(boa, "request");}baos.close();//生成ByteBufferthis.bb = ByteBuffer.wrap(baos.toByteArray());//将bytebuffer的前4个字节修改成真正的长度，总长度减去一个int的长度头 this.bb.putInt(this.bb.capacity() - 4);//准备给后续读    让buffer position = 0this.bb.rewind();} catch (IOException e) {LOG.warn("Ignoring unexpected exception", e);}}

Code 16：IO read

if (sockKey.isReadable()) {//先从Channel读4个字节，代表头  int rc = sock.read(incomingBuffer);if (rc < 0) {throw new EndOfStreamException("Unable to read additional data from server sessionid 0x"+ Long.toHexString(sessionId)+ ", likely server has closed socket");}if (!incomingBuffer.hasRemaining()) {incomingBuffer.flip();if (incomingBuffer == lenBuffer) {recvCount++;readLength();} //初始化else if (!initialized) {readConnectResult();  // 读取连接结果enableRead(); // Channel 可读if (findSendablePacket(outgoingQueue,cnxn.sendThread.clientTunneledAuthenticationInProgress()) != null) {// Since SASL authentication has completed (if client is configured to do so),// outgoing packets waiting in the outgoingQueue can now be sent.enableWrite();}lenBuffer.clear();incomingBuffer = lenBuffer;updateLastHeard();initialized = true;} else {    // 处理其他请求sendThread.readResponse(incomingBuffer);lenBuffer.clear();incomingBuffer = lenBuffer;updateLastHeard();}}}

还有一个比较关键的函数就是readResponse函数，用来消费PendingQueue,处理的消息分为三类
 ping 消息 XID=-2
 auth认证消息 XID=-4
 订阅的消息，即各种变化的通知，比如子节点变化、节点内容变化，由服务器推过来的消息，获取到这类消息或通过eventThread.queueEvent将消息推入事件队列
XID=-1

Code 17： readResponse

void readResponse(ByteBuffer incomingBuffer) throws IOException {ByteBufferInputStream bbis = new ByteBufferInputStream(incomingBuffer);BinaryInputArchive bbia = BinaryInputArchive.getArchive(bbis);ReplyHeader replyHdr = new ReplyHeader();replyHdr.deserialize(bbia, "header");if (replyHdr.getXid() == -2) {// -2 is the xid for pingsif (LOG.isDebugEnabled()) {LOG.debug("Got ping response for sessionid: 0x"+ Long.toHexString(sessionId)+ " after "+ ((System.nanoTime() - lastPingSentNs) / 1000000)+ "ms");}return;}if (replyHdr.getXid() == -4) {// -4 is the xid for AuthPacket               if(replyHdr.getErr() == KeeperException.Code.AUTHFAILED.intValue()) {state = States.AUTH_FAILED;                    eventThread.queueEvent( new WatchedEvent(Watcher.Event.EventType.None, Watcher.Event.KeeperState.AuthFailed, null) );                                      }if (LOG.isDebugEnabled()) {LOG.debug("Got auth sessionid:0x"+ Long.toHexString(sessionId));}return;}if (replyHdr.getXid() == -1) {// -1 means notificationif (LOG.isDebugEnabled()) {LOG.debug("Got notification sessionid:0x"+ Long.toHexString(sessionId));}WatcherEvent event = new WatcherEvent();event.deserialize(bbia, "response");// convert from a server path to a client pathif (chrootPath != null) {String serverPath = event.getPath();if(serverPath.compareTo(chrootPath)==0)event.setPath("/");else if (serverPath.length() > chrootPath.length())event.setPath(serverPath.substring(chrootPath.length()));else {LOG.warn("Got server path " + event.getPath()+ " which is too short for chroot path "+ chrootPath);}}WatchedEvent we = new WatchedEvent(event);if (LOG.isDebugEnabled()) {LOG.debug("Got " + we + " for sessionid 0x"+ Long.toHexString(sessionId));}//将事件加入到 event队列中eventThread.queueEvent( we );return;}

结束了IO之后就是对于事件的消费，也就是一开始图示的右半部分也是接近最后部分啦

Code 18：EventThread run:

public void run() {try {isRunning = true;while (true) {// 获取事件Object event = waitingEvents.take();if (event == eventOfDeath) {wasKilled = true;} else {//处理事件processEvent(event);}if (wasKilled)synchronized (waitingEvents) {if (waitingEvents.isEmpty()) {isRunning = false;break;}}}} catch (InterruptedException e) {LOG.error("Event thread exiting due to interruption", e);}LOG.info("EventThread shut down");}}
}

最后就是processEvent了，这个就不贴代码了（代码备注的累死了），写思路。

ProcessEvent：

processEvent 是 EventThread 处理事件核心函数，核心逻辑如下：

1、如果 event instanceof WatcherSetEventPair ，取出 pair 中的 Watchers ，逐个调用 watcher.process(pair.event)
2、否则 event 为 AsyncCallback ，根据 p.response 判断为哪种响应类型，执行响应的回调 processResult 。

Watcher 和 AsyncCallback 的区别
Watcher： Watcher 是用于监听节点，session 状态的，比如 getData 对数据节点 a 设置了 watcher ，那么当 a 的数据内容发生改变时，客户端会收到 NodeDataChanged 通知，然后进行 watcher 的回调。
AsyncCallback : AsyncCallback 是在以异步方式使用 ZooKeeper API 时，用于处理返回结果的。例如：getData 同步调用的版本是： byte[] getData(String path, boolean watch,Stat stat) ，异步调用的版本是： void getData(String path,Watcher watcher,AsyncCallback.DataCallback cb,Object ctx) ，可以看到，前者是直接返回获取的结果，后者是通过 AsyncCallback 回调处理结果的。

**接下来就是客户端发送指令与负责端进行交互比如:
Ls、getChildren、getData等**

参考文献：
[1] http://www.cnblogs.com/davidwang456/p/5000927.html
[2] http://www.verydemo.com/demo_c89_i33659.html
[3] http://blog.csdn.net/pwlazy/article/details/8000566
[4] http://www.cnblogs.com/ggjucheng/p/3376548.html
[5] http://zookeeper.apache.org/doc/r3.3.6/api/index.html
[6] http://www.tuicool.com/articles/i6vMVze
[7]http://www.ibm.com/developerworks/cn/opensource/os-cn-apache-zookeeper-watcher/

Zookeeper 客户端源码吐血总结相关推荐

ZooKeeper客户端源码（零）——客户端API使用
首发CSDN:徐同学呀,原创不易,转载请注明源链接.我是徐同学,用心输出高质量文章,希望对你有所帮助. 本篇源码基于ZooKeeper3.7.0版本. 一.建立连接和会话客户端可以通过创建一个 Zo ...
ZooKeeper客户端源码（三）——Watcher注册与通知
首发CSDN:徐同学呀,原创不易,转载请注明源链接.我是徐同学,用心输出高质量文章,希望对你有所帮助. 本篇源码基于ZooKeeper3.7.0版本. ZooKeeper 提供了分布式数据的发布/订阅 ...
ZooKeeper客户端源码（一）——向服务端建立连接+会话建立+心跳保持长连接
首发CSDN:徐同学呀,原创不易,转载请注明源链接.我是徐同学,用心输出高质量文章,希望对你有所帮助. 一.从ZooKeeper实例初始化开始 ZooKeeper 提供了原生的客户端库,虽然不好用,但 ...
zookeeper 客户端_zookeeper进阶-客户端源码详解
流程图先看一下客户端源码的流程图总体流程总体流程开启SendThread线程开启EventThread 总结下面根据源码讲解,大家整合源码和流程图一起看最好,本篇内容比较多建议收藏起来看. ...
boost::asio异步模式的C/S客户端源码实现
异步模式的服务器源码 //g++ -g async_tcp_server.cpp -o async_tcp_server -lboost_system //#include <iostream& ...
grpc-go客户端源码分析
grpc-go客户端源码分析代码讲解基于v1.37.0版本. 和grpc-go服务端源码分析一样,我们先看一段示例代码, const (address = "localhost:50051 ...
RabbitMQ初步到精通-第十章-RabbitMQ之Spring客户端源码
目录第十章-RabbitMQ之Spring客户端源码 1. 前言 2. 客户端消费代码 2.1 消费的实现方式 2.2 消费中注解解释 2.3 推测Spring实现过程 3.MQ消费源码分析 3.1 ...
WordPress Blog Android客户端源码分析（一）
一直想找一个大型的Android开源项目进行分析,由于自身和导师课程需要选择了wordpress的Android客户端源码进行学习和解读.源码github官方下载地址:开源项目地址.分析源码的最佳手段 ...
使用live555客户端源码遇到的问题及解决方法
使用live555客户端源码拉rtsp流遇到两个问题,正常测试拉取海康摄像头没问题: 1.拉有些厂商的rtsp流会间隔一段时间断开连接: 2.与大华摄像头建立连接时,发送DESCRIBE命令后很长时间 ...

Zookeeper 客户端源码吐血总结

Zookeeper 客户端源码吐血总结相关推荐

最新文章

热门文章