AndroidPN客户端的阻塞读写(1)

      androidPN服务端用的是mina,略去不表，客户端的socket通讯用的是asmack,期间使用xmpp协议通讯，这个xmpp通用是通用了，但用的是xml格式互发，之间不得不加了一堆的xml解析，大部分篇幅都是干这个，对此没多大兴趣，这里只是简单记录一下阅读源码中client与server的阻塞读写，寻找可以借鉴之处。

     客户端启动之后，负责管理连接的XMPPConnection初始化：

 if (isFirstInitialization) {
                packetWriter = new PacketWriter(this);
                packetReader = new PacketReader(this);

   分别负责读写。然后启动二者：

            // Start the packet writer. This will open a XMPP stream to the server
            packetWriter.startup();
            // Start the packet reader. The startup() method will block until we
            // get an opening stream packet back from server.
            packetReader.startup();
            // Make note of the fact that we're now connected.
            connected = true;

   

先来看packetWriter是如何向服务器发送数据的，它的初始化方法

    protected void init() {
        this.writer = connection.writer;
        done = false;
        writerThread = new Thread() {
            public void run() {
                writePackets(this);
            }
        };
        writerThread.setName("Smack Packet Writer (" + connection.connectionCounterValue + ")");
        writerThread.setDaemon(true);
    }

开一个守候线程writeThread,跑writePackets(this),该方法主要代码：

            // Write out packets from the queue.
            while (!done && (writerThread == thisThread)) {
                Packet packet = nextPacket();
                if (packet != null) {
                    writer.write(packet.toXML());
                    if (queue.isEmpty()) {
                        writer.flush();
                    }
                }
            }

 其中queue,线程安全:

private final BlockingQueue<Packet> queue;

 queue中有数据就write到服务器，如果没数据，阻塞在nextPacket()：

    private Packet nextPacket() {
        Packet packet = null;
        // Wait until there's a packet or we're done.
        while (!done && (packet = queue.poll()) == null) {
            try {
                synchronized (queue) {
                    queue.wait();
                }
            }
            catch (InterruptedException ie) {
                // Do nothing
            }
        }
        return packet;
    }

 看看，queue.wait(),写线程阻塞于此，省电! 既然有wait(),必然有notifyAll():

    public void sendPacket(Packet packet) {
        if (!done) {
            // Invoke interceptors for the new packet that is about to be sent. Interceptors
            // may modify the content of the packet.
            connection.firePacketInterceptors(packet);

            try {
                queue.put(packet);
            }
            catch (InterruptedException ie) {
                ie.printStackTrace();
                return;
            }
            synchronized (queue) {
                queue.notifyAll();
            }

            // Process packet writer listeners. Note that we're using the sending
            // thread so it's expected that listeners are fast.
            connection.firePacketSendingListeners(packet);
        }
    }

 当其他线程把要写的数据put进queue时，writerThread即被唤醒,继续运作，标准的生产消费模式。这个过程还是比较简单的，一目了然，相比之下PacketReader就不那么省心。

 

    PacketReader初始化：

    protected void init() {
//...
        readerThread = new Thread() {
            public void run() {
                parsePackets(this);
            }
        };
 //...
        resetParser();
    }

 接着看parsePackets:

    private void parsePackets(Thread thread) {
        try {
            int eventType = parser.getEventType();
            do {
                if (eventType == XmlPullParser.START_TAG) {
//...
//...很长很长，都在xml纠结
//...
                eventType = parser.next();
            } while (!done && eventType != XmlPullParser.END_DOCUMENT && thread == readerThread);
        }

好吧，循环体是有了，但没见着一个reader.read()，阻塞读在哪里？还有输入流在哪里？

注意到上面初始化方法中最后resetParser()：

    private void resetParser() {
        try {
            parser = XmlPullParserFactory.newInstance().newPullParser();
            parser.setFeature(XmlPullParser.FEATURE_PROCESS_NAMESPACES, true);
            parser.setInput(connection.reader);
        }
        catch (XmlPullParserException xppe) {
            xppe.printStackTrace();
        }
    }

 这里把输入流connection.reader交给了parser。看来只能从parser中查找，XmlPullParserFactory中

    public XmlPullParser newPullParser() throws XmlPullParserException {
        final XmlPullParser pp = new KXmlParser();
        for (Map.Entry<String, Boolean> entry : features.entrySet()) {
            pp.setFeature(entry.getKey(), entry.getValue());
        }

        return pp;
    }

 parser来自于org.kxml2.io.KXmlParser，继续下kxml2源码包找，真够折腾的

  public void setInput(Reader reader) throws XmlPullParserException {
        this.reader = reader;
//...
        if (reader == null)
            return;
//...
    }

 可见socket输入流交给了parser,顺便看下parsePackets方法中用到的parser.next()等方法，都调用了peek()方法：

    /** Does never read more than needed */

    private final int peek(int pos) throws IOException {
        while (pos >= peekCount) {
            int nw;
            if (srcBuf.length <= 1)
                nw = reader.read();
            else if (srcPos < srcCount)
                nw = srcBuf[srcPos++];
            else {
                srcCount = reader.read(srcBuf, 0, srcBuf.length);
                if (srcCount <= 0)
                    nw = -1;
                else
                    nw = srcBuf[0];
                srcPos = 1;
            }
            if (nw == '\r') {
                wasCR = true;
                peek[peekCount++] = '\n';
            }
            else {
                if (nw == '\n') {
                    if (!wasCR)
                        peek[peekCount++] = '\n';
                }
                else
                    peek[peekCount++] = nw;
                wasCR = false;
            }
        }
        return peek[pos];
    }

 这里终于出现了reader.read()，就在这里阻塞读。这是一个多么苦逼的过程。为了xml解析不得不把一行代码化为数百行代码，而且这样很费电的你知道吗^_^。

 到此结束，化了一上午看这些，有些收获,有些头痛