概述
Smack是一個開源的實現(xiàn)了XMPP協(xié)議的庫苟翻，特別是4.1.0版本以后，直接支持Android系統(tǒng)骗污，無需再使用以前那個專門針對Android系統(tǒng)的aSmack移植庫了.雖然在移動端上崇猫，用XMPP協(xié)議來做IM并不是一個最優(yōu)選擇，市面上這些大公司基本都是用自己定制的私有協(xié)議需忿，沒有采用XMPP協(xié)議的诅炉，不過我們可以拋開協(xié)議層面，只分析一下Smack庫在網(wǎng)絡層的實現(xiàn)屋厘，也是有借鑒意義的涕烧。
總體結構
Smack抽象出一個XMPPConnection的概念，要想收發(fā)消息汗洒，首先得建立這個connection议纯，而且這種connection是可以由多個實例的。XMPPConnection只是一個接口溢谤，AbstractXMPPConnection實現(xiàn)了這個接口并加入了login瞻凤，connect，processStanza等方法溯香。AbstractXMPPConnection有兩個實現(xiàn)類鲫构，XMPPBOSHConnection和XMPPTCPConnection。其中XMPPBOSHConnection是基于Http協(xié)議來實現(xiàn)的玫坛，而XMPPTCPConnection是直接用Socket來實現(xiàn)的長連接通信，本文分析的也就是XMPPTCPConnection包晰。一個簡單的使用實例如下：
XMPPTCPConnection con = new XMPPTCPConnection("igniterealtime.org");
// Connect to the server
con.connect();
// Most servers require you to login before performing other tasks.
con.login("jsmith", "mypass");
// Start a new conversation with John Doe and send him a message.
Chat chat = ChatManager.getInstanceFor(con).createChat("jdoe@igniterealtime.org", new MessageListener() {
public void processMessage(Chat chat, Message message) {
// Print out any messages we get back to standard out.
System.out.println("Received message: " + message);
}
});
chat.sendMessage("Howdy!");
// Disconnect from the server
con.disconnect();
接口介紹
XMPPConnection這個接口里有幾個主要的方法 :
public void sendStanza(Stanza stanza) throws NotConnectedException, InterruptedException;
public void addConnectionListener(ConnectionListener connectionListener);
public void addPacketInterceptor(StanzaListener packetInterceptor, StanzaFilter packetFilter);
public void addPacketSendingListener(StanzaListener packetListener, StanzaFilter packetFilter);
public PacketCollector createPacketCollector(StanzaFilter packetFilter);
public void addAsyncStanzaListener(StanzaListener packetListener, StanzaFilter packetFilter);
public void addSyncStanzaListener(StanzaListener packetListener, StanzaFilter packetFilter);
sendStanza 發(fā)送包到服務器湿镀。在最新版的Smack中炕吸，Stanza就是以前版本中的Packet
addConnectionListener 添加ConnectionListener到XMPPConnection中。在該Listener中勉痴，監(jiān)聽者可以得到連接是否成功建立赫模，連接關閉，連接異常關閉蒸矛，重連是否成功等事件
addPacketInterceptor 向Connection中注冊攔截器StanzaListener瀑罗，所有發(fā)往服務器的包都會先過一遍攔截器，你可以在攔截器中對這些包進行處理雏掠；StanzaFilter過濾器可以允許你定制哪些包才需要攔截; StanzaListener和StanzaFilter常常配對使用斩祭，代碼中有各種wrapper類（如ListenerWrapper、InterceptorWrapper等）乡话，就是把這兩個接口組合在一個類中摧玫，一個負責過濾包，一個負責實際處理包
addPacketSendingListener 注冊一個Listener绑青，當把包通過Socket寫出去后诬像，會回調(diào)這個Listener告知正在發(fā)送狀態(tài)
createPacketCollector 當你想接收某種類型的包時，可以新建一個包收集器闸婴。和StanzaListener不同坏挠，包收集器是阻塞式的，直到指定的包收到或者出現(xiàn)超時（我們可以設置等待一個包的最大時間）等異常
PacketCollector messageCollector = connection.createPacketCollector(messageFilter);
try {
connection.createPacketCollectorAndSend(request).nextResultOrThrow();
// Collect the received offline messages
Message message = messageCollector.nextResult();
while (message != null) {
messages.add(message);
message = messageCollector.nextResult();
}
}
finally {
// Stop queuing offline messages
messageCollector.cancel();
}
return messages;
- addAsyncStanzaListener和addSyncStanzaListener 添加處理收到的包的回調(diào)接口邪乍；其中一個叫同步一個叫異步區(qū)別在于,執(zhí)行回調(diào)方法所用的線程池不一樣癞揉，其中異步用的是Executors.newCachedThreadPool，而同步用的是一個Executors.newSingleThreadExecutor溺欧，可以保證執(zhí)行順序
// First handle the async recv listeners. Note that this code is very similar to what follows a few lines below,
// the only difference is that asyncRecvListeners is used here and that the packet listeners are started in
// their own thread.
final Collection<StanzaListener> listenersToNotify = new LinkedList<StanzaListener>();
synchronized (asyncRecvListeners) {
for (ListenerWrapper listenerWrapper : asyncRecvListeners.values()) {
if (listenerWrapper.filterMatches(packet)) {
listenersToNotify.add(listenerWrapper.getListener());
}
}
}
for (final StanzaListener listener : listenersToNotify) {
asyncGo(new Runnable() {
@Override
public void run() {
try {
listener.processPacket(packet);
} catch (Exception e) {
LOGGER.log(Level.SEVERE, "Exception in async packet listener", e);
}
}
});
}
// Loop through all collectors and notify the appropriate ones.
for (PacketCollector collector: collectors) {
collector.processPacket(packet);
}
// Notify the receive listeners interested in the packet
listenersToNotify.clear();
synchronized (syncRecvListeners) {
for (ListenerWrapper listenerWrapper : syncRecvListeners.values()) {
if (listenerWrapper.filterMatches(packet)) {
listenersToNotify.add(listenerWrapper.getListener());
}
}
}
// Decouple incoming stanza processing from listener invocation. Unlike async listeners, this uses a single
// threaded executor service and therefore keeps the order.
singleThreadedExecutorService.execute(new Runnable() {
@Override
public void run() {
for (StanzaListener listener : listenersToNotify) {
try {
listener.processPacket(packet);
} catch(NotConnectedException e) {
LOGGER.log(Level.WARNING, "Got not connected exception, aborting", e);
break;
} catch (Exception e) {
LOGGER.log(Level.SEVERE, "Exception in packet listener", e);
}
}
}
});
AbstractXMPPConnection實現(xiàn)了XMPPConnection接口喊熟，各種Listener的注冊和回調(diào)就是在這個類里完成的，但如login姐刁，connect芥牌，shutdown等方法的具體實現(xiàn)是位于其子類中的。
連接過程
真正執(zhí)行連接動作的是XMPPTCPConnection中connectInternal的方法
protected void connectInternal() throws SmackException, IOException, XMPPException, InterruptedException {
closingStreamReceived.init();
// Establishes the TCP connection to the server and does setup the reader and writer. Throws an exception if
// there is an error establishing the connection
connectUsingConfiguration();
// We connected successfully to the servers TCP port
initConnection();
// Wait with SASL auth until the SASL mechanisms have been received
saslFeatureReceived.checkIfSuccessOrWaitOrThrow();
// Make note of the fact that we're now connected.
connected = true;
callConnectionConnectedListener();
}
connectUsingConfiguration方法中聂使，用配置類XMPPTCPConnectionConfiguration提供的hostAddress壁拉，timeout等數(shù)據(jù)創(chuàng)建一個Socket連接出來。隨后進行了一些初始化柏靶，例如初始化reader弃理，writer變量：
private void initReaderAndWriter() throws IOException {
InputStream is = socket.getInputStream();
OutputStream os = socket.getOutputStream();
if (compressionHandler != null) {
is = compressionHandler.getInputStream(is);
os = compressionHandler.getOutputStream(os);
}
// OutputStreamWriter is already buffered, no need to wrap it into a BufferedWriter
writer = new OutputStreamWriter(os, "UTF-8");
reader = new BufferedReader(new InputStreamReader(is, "UTF-8"));
// If debugging is enabled, we open a window and write out all network traffic.
initDebugger();
}
PacketWriter對包的發(fā)送進行了封裝，該類里維護一個BlockingQueue屎蜓，所有要發(fā)送的包都先插入到這個隊列中痘昌，同時起一個線程不停消費這個隊列，最終是通過writer把數(shù)據(jù)寫往服務器
while (!queue.isEmpty()) {
Element packet = queue.remove();
writer.write(packet.toXML().toString());
}
writer.flush();
而PacketReader則是對包的讀取和解析進行了封裝，類里面有個XmlPullParser辆苔，通過reader進行了初始化
packetReader.parser = PacketParserUtils.newXmppParser(reader);
然后起了一個線程不停進行包的解析
Async.go(new Runnable() {
public void run() {
parsePackets();
}
}, "Smack Packet Reader (" + getConnectionCounter() + ")");
}
解析出來的包回調(diào)到AbstractXMPPConnection類中的parseAndProcessStanza方法算灸，最終調(diào)用各種已注冊好的StanzaListener、PacketCollector來處理
XMPPConnectionRegistry
這個靜態(tài)類中有個ConnectionCreationListener的集合
private final static Set<ConnectionCreationListener> connectionEstablishedListeners =
new CopyOnWriteArraySet<ConnectionCreationListener>();
當XMPPConnection初始化的時候驻啤，會通知給各個Listener
protected AbstractXMPPConnection(ConnectionConfiguration configuration) {
saslAuthentication = new SASLAuthentication(this, configuration);
config = configuration;
// Notify listeners that a new connection has been established
for (ConnectionCreationListener listener : XMPPConnectionRegistry.getConnectionCreationListeners()) {
listener.connectionCreated(this);
}
}
像ReconnectionManager菲驴，PingManager等策略管理類，會在靜態(tài)代碼塊中直接注冊ConnectionCreationListener
static {
XMPPConnectionRegistry.addConnectionCreationListener(new ConnectionCreationListener() {
public void connectionCreated(XMPPConnection connection) {
if (connection instanceof AbstractXMPPConnection) {
ReconnectionManager.getInstanceFor((AbstractXMPPConnection) connection);
}
}
});
}
ReconnectionManager
由于可以創(chuàng)建多個XMPPConnection的實例骑冗，ReconnectionManager的實例也有多個赊瞬，和XMPPConnection一一對應，實際上ReconnectionManager持有了XMPPConnection的弱引用贼涩，用于進行與Connection相關的操作巧涧。
類里面還定義了不同的重連策略ReconnectionPolicy，有按固定頻率重連的磁携，也有按隨機間隔重連的褒侧，
private int timeDelay() {
attempts++;
// Delay variable to be assigned
int delay;
switch (reconnectionPolicy) {
case FIXED_DELAY:
delay = fixedDelay;
break;
case RANDOM_INCREASING_DELAY:
if (attempts > 13) {
delay = randomBase * 6 * 5; // between 2.5 and 7.5 minutes (~5 minutes)
}
else if (attempts > 7) {
delay = randomBase * 6; // between 30 and 90 seconds (~1 minutes)
}
else {
delay = randomBase; // 10 seconds
}
break;
default:
throw new AssertionError("Unknown reconnection policy " + reconnectionPolicy);
}
return delay;
}
ReconnectionManager向XMPPConnection注冊了ConnectionListener，當XMPPConnection中發(fā)生連接異常時谊迄，如PacketWriter闷供、PacketReader讀寫包異常時，會通過ConnectionListener中的connectionClosedOnError方法统诺，通知ReconnectionManager進行重連重試歪脏。
PingManager、ServerPingWithAlarmManager
PingManager實現(xiàn)了協(xié)議規(guī)定的定時發(fā)送Ping消息到服務器的策略粮呢，默認是30分鐘的間隔婿失。ServerPingWithAlarmManager是針對Android平臺的實現(xiàn)，用AlarmManager來實現(xiàn)的定時策略啄寡，在代碼里寫死是30分鐘的頻率豪硅，這在移動端肯定是不適用的，另外也沒看到針對各種網(wǎng)絡環(huán)境的處理挺物，看來為保證長連接的穩(wěn)定性懒浮，需要開發(fā)者自己再去實現(xiàn)一些心跳和重連策略
