Mosquitto 是一個IBM 開源pub/sub訂閱發布協議 MQTT 的一個單機版實現（目前也只有單機版），MQTT主打輕便，比較適用於移動設備等上面，花費流量少，解析代價低。相對於XMPP等來說，簡單許多。
MQTT采用二進制協議，而不是XMPP的XML協議，所以一般消息甚至只需要花費2個字節的大小就可以交換信息了，對於移動開發比較有優勢。
IBM雖然開源了其MQTT消息協議，但是卻沒有開源其RSMB服務端程序，不過還好目前有比較穩定的實現可用，本文的Mosquitto是其中比較活躍的實現之一，具體在 這里 有目前的實現列表可供選擇。
趁着大腦還沒有進入睡眠狀態記錄一下剛才看代碼學到的東西。我下載的版本是1.2.2版，在 這里 可以找到 下載鏈接 。
零、介紹
關於 MQTT 3.1協議 本身比較簡單，42頁的PDF介紹完了，相比XMPP那長長的文檔，謝天謝地了。由於剛看，所以很多細節都沒有深入進去，這里只是記錄個大概，后續有時間慢慢補好坑吧。
總體來說，mosquitto實現有如下幾個特點：
- poll()異步模型，竟然不是epoll，也許這注定了其只能支持十幾萬連接同時在線的悲劇吧。
- 內存處理方面幾乎沒有任何優化，但簡單可依賴；
- 多線程程序，許多地方都得加鎖訪問。但其實多線程的需求沒那么強烈，可以考慮避免；
總之，是一個比較簡單單可以適用於一般的服務中提供pub/sub功能支持，但如果放到大量並發的系統中，可以優化的地方還有很多。關於mosquitto的性能，暫時沒有找到官方的評測，不過在 郵件組里面找到的一些討論 似乎顯示其性能上限為20W連接時在線的狀態，當然具體取決於業務邏輯，交互是否很多等。不過這樣的成績還是不錯的。一台機器可以起多個實例的嘛。
一、初始化
mosquitto.c文件main開頭調用_mosquitto_net_init初始化SSL加密的庫，然后調用mqtt3_config_init初始化配置的各個數據結構為默認值。配置文件的解析由mqtt3_config_parse_args牽頭完成，具體配置文件解析就不多寫了，fgets一行行的讀取配置，然后設置到config全局變量中。其中包括對於監聽地址等的讀取。
然后保存pid進程號。mqtt3_db_open打開db文件
int main(int argc, char *argv[]) { memset(&int_db, 0, sizeof(struct mosquitto_db)); _mosquitto_net_init(); mqtt3_config_init(&config); rc = mqtt3_config_parse_args(&config, argc, argv);//k: init && load config file, set struct members
配置讀取完后，就可以打開監聽端口了，使用mqtt3_socket_listen打開監聽端口，並將SOCK套接字放在局部變量listensock里面，以便后面統一使用。
listener_max = -1; listensock_index = 0; for(i=0; i<config.listener_count; i++){ if(mqtt3_socket_listen(&config.listeners[i])){ _mosquitto_free(int_db.contexts); mqtt3_db_close(&int_db); if(config.pid_file){ remove(config.pid_file); } return 1; } listensock_count += config.listeners[i].sock_count; listensock = _mosquitto_realloc(listensock, sizeof(int)*listensock_count); if(!listensock){ _mosquitto_free(int_db.contexts); mqtt3_db_close(&int_db); if(config.pid_file){ remove(config.pid_file); } return 1; } for(j=0; j<config.listeners[i].sock_count; j++){ if(config.listeners[i].socks[j] == INVALID_SOCKET){ _mosquitto_free(int_db.contexts); mqtt3_db_close(&int_db); if(config.pid_file){ remove(config.pid_file); } return 1; } listensock[listensock_index] = config.listeners[i].socks[j]; if(listensock[listensock_index] > listener_max){ listener_max = listensock[listensock_index]; } listensock_index++; } }
關於mqtt3_socket_listen函數也比較經典，socket(),bind(), listen()的流程，不同的是使用了新版的套接字信息獲取函數getaddrinfo，該函數支持IPV4和IPV6，對應用層透明，不需要處理這些信息。
mqtt3_socket_listen(struct _mqtt3_listener *listener) { snprintf(service, 10, "%d", listener->port); memset(&hints, 0, sizeof(struct addrinfo)); hints.ai_family = PF_UNSPEC; hints.ai_flags = AI_PASSIVE; hints.ai_socktype = SOCK_STREAM; //導致下面返回多個鏈表節的的因素可能有： //hostname參數關聯的地址有多個，那么每個返回一個節點；比如host為域名的時候，nslookup返回幾個ip就有幾個 //service參數指定的服務會吃多個套接字接口類型，那么也返回多個 if(getaddrinfo(listener->host, service, &hints, &ainfo)) return INVALID_SOCKET; listener->sock_count = 0; listener->socks = NULL; for(rp = ainfo; rp; rp = rp->ai_next){ //···· sock = socket(rp->ai_family, rp->ai_socktype, rp->ai_protocol); if(sock == -1){ strerror_r(errno, err, 256); _mosquitto_log_printf(NULL, MOSQ_LOG_WARNING, "Warning: %s", err); continue; } listener->sock_count++; listener->socks = _mosquitto_realloc(listener->socks, sizeof(int)*listener->sock_count); if(!listener->socks){ _mosquitto_log_printf(NULL, MOSQ_LOG_ERR, "Error: Out of memory."); return MOSQ_ERR_NOMEM; } listener->socks[listener->sock_count-1] = sock; /* Set non-blocking */ opt = fcntl(sock, F_GETFL, 0); if(bind(sock, rp->ai_addr, rp->ai_addrlen) == -1){ strerror_r(errno, err, 256); _mosquitto_log_printf(NULL, MOSQ_LOG_ERR, "Error: %s", err); COMPAT_CLOSE(sock); return 1; } if(listen(sock, 100) == -1){ strerror_r(errno, err, 256); _mosquitto_log_printf(NULL, MOSQ_LOG_ERR, "Error: %s", err); COMPAT_CLOSE(sock); return 1; } } freeaddrinfo(ainfo); }
二、消息事件循環
打開監聽套接字后，就可以進入消息事件循環，標准網絡服務程序的必須過程。這個由main函數調用mosquitto_main_loop啟動。mosquitto_main_loop函數主體也是一個大循環，在循環里面進行超時檢測，事件處理，網絡讀寫等等。由於使用poll模型，所以就需要在進行poll()等待之前准備需要監聽的套接字數組列表pollfds，效率低的地方就在這里。
對於監聽套接字，簡單將其加入pollfds里面，注冊POLLIN可讀事件即可。如果對於其他跟客戶端等的連接，就需要多做一步操作了。如果是橋接模式，進行相應的處理，這里暫時不介紹橋接模式，橋接模式是為了分布式部署加入的非標准協議，目前只有IBM rsmb和mosquitto實現了。
對於跟客戶端的連接，mosquitto會在poll等待之前調用mqtt3_db_message_write嘗試發送一次未發送的數據給對方，避免不必要的等待可能。
int mosquitto_main_loop(struct mosquitto_db *db, int *listensock, int listensock_count, int listener_max) { memset(pollfds, -1, sizeof(struct pollfd)*pollfd_count); pollfd_index = 0; for(i=0; i<listensock_count; i++){//注冊監聽sock的pollfd可讀事件。也就是新連接事件 pollfds[pollfd_index].fd = listensock[i]; pollfds[pollfd_index].events = POLLIN; pollfds[pollfd_index].revents = 0; pollfd_index++; } time_count = 0; for(i=0; i<db->context_count; i++){//遍歷每一個客戶端連接,嘗試將其加入poll數組中 if(db->contexts[i]){ //···· /* Local bridges never time out in this fashion. */ if(!(db->contexts[i]->keepalive) || db->contexts[i]->bridge || now - db->contexts[i]->last_msg_in < (time_t)(db->contexts[i]->keepalive)*3/2){ //在進入poll等待之前，先嘗試將未發送的數據發送出去 if(mqtt3_db_message_write(db->contexts[i]) == MOSQ_ERR_SUCCESS){ pollfds[pollfd_index].fd = db->contexts[i]->sock; pollfds[pollfd_index].events = POLLIN | POLLRDHUP; pollfds[pollfd_index].revents = 0; if(db->contexts[i]->current_out_packet){ pollfds[pollfd_index].events |= POLLOUT; } db->contexts[i]->pollfd_index = pollfd_index; pollfd_index++; }else{//嘗試發送失敗，連接出問題了 mqtt3_context_disconnect(db, db->contexts[i]); } }else{//超過1.5倍的時間，超時關閉連接 if(db->config->connection_messages == true){ _mosquitto_log_printf(NULL, MOSQ_LOG_NOTICE, "Client %s has exceeded timeout, disconnecting.", db->co ntexts[i]->id); } /* Client has exceeded keepalive*1.5 */ mqtt3_context_disconnect(db, db->contexts[i]);//關閉連接，清空數據，后續還可以用.sock=INVALID_SOCKET } }else{ #endif if(db->contexts[i]->clean_session == true){ //這個連接上次由於什么原因，掛了，設置了clean session，所以這里直接徹底清空其結構 mqtt3_context_cleanup(db, db->contexts[i], true); db->contexts[i] = NULL; }else if(db->config->persistent_client_expiration > 0){ //協議規定persistent_client的狀態必須永久保存，這里避免連接永遠無法刪除，增加這個超時選項。 //也就是如果一個客戶端斷開連接一段時間了，那么我們會主動干掉他 /* This is a persistent client, check to see if the * last time it connected was longer than * persistent_client_expiration seconds ago. If so, * expire it and clean up. */ if(now > db->contexts[i]->disconnect_t+db->config->persistent_client_expiration){ _mosquitto_log_printf(NULL, MOSQ_LOG_NOTICE, "Expiring persistent client %s due to timeout.", db- >contexts[i]->id); #ifdef WITH_SYS_TREE g_clients_expired++; #endif db->contexts[i]->clean_session = true; mqtt3_context_cleanup(db, db->contexts[i], true); db->contexts[i] = NULL; } } #ifdef WITH_BRIDGE }
然后先使用mqtt3_db_message_timeout_check檢測一下超時沒有收到客戶端回包確認的消息，mosquitto對於超時的消息處理，是會進行重發的。不過理論上，TCP是不需要重發的，具體見這里： MQTT消息推送協議應用數據包超時是否需要重發？ 不過，由於mosquitto對於客戶端斷開連接的處理比較弱，連接重新建立后，使用的相關數據結構還是相同的，因此重發其實也可以，只是這個時候的重發，實際上是在一個連接上沒有收到ACK回包，然后后續建立的新連接上進行重傳。不是在一個連接上重傳。但是這樣其實也有很多弊端，比如客戶端必須支持消息的持久化記錄，否則容易雙方對不上話的情況。
int mqtt3_db_message_timeout_check(struct mosquitto_db *db, unsigned int timeout) {//循環遍歷每一個連接的每個消息msg,看起是否超時，如果超時，將消息狀態改為上一個狀態，從而后續觸發重發 int i; time_t threshold; enum mosquitto_msg_state new_state = mosq_ms_invalid; struct mosquitto *context; struct mosquitto_client_msg *msg; threshold = mosquitto_time() - timeout; for(i=0; i<db->context_count; i++){//遍歷每一個連接， context = db->contexts[i]; if(!context) continue; msg = context->msgs; while(msg){//遍歷每個msg消息，看看其狀態，如果超時了，那么從上一個消息開始重發.其實不需要重發http://chenzhenianqing.cn/ar ticles/977.html //當然如果這個是復用了之前斷開過的連接，那就需要重發。但是，這個時候其實可以重發整個消息的。不然容易出問題，客戶端難> 度大 if(msg->timestamp < threshold && msg->state != mosq_ms_queued){ switch(msg->state){ case mosq_ms_wait_for_puback: new_state = mosq_ms_publish_qos1; break; case mosq_ms_wait_for_pubrec: new_state = mosq_ms_publish_qos2; break; case mosq_ms_wait_for_pubrel: new_state = mosq_ms_send_pubrec; break; case mosq_ms_wait_for_pubcomp: new_state = mosq_ms_resend_pubrel; break; default: break; } if(new_state != mosq_ms_invalid){ msg->timestamp = mosquitto_time();//設置當前時間，下次依據來判斷超時
超時提前檢測完成后就可以進入poll等待了。等待完成后，對於有可讀事件的連接，調用loop_handle_reads_writes進行事件讀寫處理，對於監聽端口的事件，使用mqtt3_socket_accept去接受新連接。
loop_handle_reads_writes新事件處理函數比較簡單，主體還是循環判斷可讀可寫事件，進行相應的處理。具體不多介紹了，需要關注的是由於是異步讀寫，所以需要記錄上次讀寫狀態，以便下次進入上下午繼續讀取數據。可寫事件由_mosquitto_packet_write完成，可讀事件由_mosquitto_packet_read完成。
新客戶端連接的事件則由qtt3_socket_accept完成，其會將新連接放在db->contexts[i]數組的某個空位置，每次都會遍歷尋找一個空的槽位放新連接。這里有個小優化其實就是用hints的機制，記錄上次的查找位置，避免多次重復的從前面找到后面。
連接讀寫事件處理完成后，mosquitto會檢測是否需要重新reload部分配置文件。這個由SIGHUP的信號觸發。
限於篇幅，具體的邏輯請求處理下次再介紹了。
三、總結
mosquitto是一個簡單可依賴的開源MQTT實現，能支持10W左右的同時在線（未親測），單機版本，但通過bridge橋接模式支持部分分布式，但有限；協議本身非常適合在移動設備上使用，耗電少，處理快，屬於header上帶有消息體長度的協議，這個在異步網絡事件代碼編寫時是碼農最愛的，哈哈。
對於后續的提高優化的地方，簡單記錄幾點：
- 發送數據用writev
- poll -> epoll ，用以支持更高的冰法；
- 改為單線程版本，降低鎖開銷，目前鎖開銷還是非常大的。目測可以改為單進程版本，類似redis，精心維護的話應該能達到不錯的效果；
- 網絡數據讀寫使用一次盡量多讀的方式，避免多次進入系統調用；
- 內存操作優化。不free，留着下次用；
- 考慮使用spwan-fcgi的形式或者內置一次啟動多個實例監聽同一個端口。這樣能更好的發揮機器性能，達到更高的性能；
參考：