千萬級郵件系統的設計及數據庫中間件部分實現
(作者：藝龍公司技術研發經理 高鵬)
近日，藝龍公司推出了“藝龍五星級企業郵箱系統”和“藝龍個人5秒郵”的收費郵件系統，這兩套系統都采用的是藝龍公司與全世界最大的郵件運營商mail.com合作設計和開發的千萬級郵件系統：“ComConnect”。筆者是該系統的主要設計人員，本文將介紹這套系統的技術特色，以及數據庫中間件部分實現。 藝龍公司的“ComConnect”系統可以運行在大部分的unix系統平臺，包括solaris、linux等，采用模塊化的設計，整個系統分為下面八大模塊： 1．數據庫中間件模塊＋緩存模塊 這是整套系統的主線，貫穿于系統的其他所有模塊。 在大規模郵件系統中，為了管理的方便和系統的分布式部署，都需要采用SQL數據庫保存一些信息，例如帳號信息等。例如，當用戶通過POP3登陸時，服務器端會啟動一個程序，與SQL數據庫連接，并查詢該用戶身份的合法性，當用戶退出時，該程序會斷開與SQL數據庫的連接。這樣，大量用戶的登陸、退出，都會對應于SQL數據庫的連接、斷開操作，降低了系統的效率，也增加了SQL數據庫的負載。
在ComConnect系統中，所有數據庫的操作，都是通過“數據庫中間件”模塊完成的。該模塊會保持與SQL數據庫的連接，永不斷開。當有程序需要查詢數據庫時，會通過unix domain socket與本地的數據庫中間件服務器相連接，中間件會取得查詢需求，并進行實際的SQL查詢，并將結果通過unix domain socket返回給應用程序。同時，對于所得的查詢，中間件服務器會進行本地緩存，保存到內存數據庫BerkeleyDB3(http://www.sleepycat.com)中，這樣下次同樣的查詢請求，就不必SQL查詢了，因為結果已經被保存在本地緩存了。這種設計，一方面，避免了頻繁的SQL數據庫的連接、斷開的操作，另一方面，由于本地緩存的存在，也大大降低了SQL數據庫的負載，提高了系統的響應效率。
2．收信模塊
該模塊對應于DNS中的MX記錄。
傳統的收信模塊運行方式是，在得到一封郵件后，保存到本地，或通過SMTP協議進行轉發。而在ComConnect系統中，采用LMTP(local mail transport protocol)協議連接“收信模塊”和“郵件存儲模塊”，在這兩個模塊之間，永久保持若干個通道，當收信模塊得到新的郵件時，就會利用這些已經存在的通道將郵件內容傳輸給后臺的郵件存儲模塊。這種設計，大大減少了SMTP服務器頻繁的fork、exit過程，從而提高了效率。 3．發信模塊 該模塊是整個系統中最獨立的模塊。為了提高安全性，ComConnect系統通過cyrus-sasl實現了發信服務器的用戶身份認證功能。認證部分通過“認證模塊”完成。 4．中央數據庫模塊
與其他大規模郵件系統一樣，ComConnect系統也需要中央數據庫(支持SQL標準)模塊。但與其他系統不同的是，ComConnect系統的中央數據庫保存的內容非常少，僅僅保存用戶帳號信息。而且，由于“數據庫中間件”和“緩存”模塊的存在，在理想狀態下，中央數據庫根本不需要存在。
這種設計，極大地提高了系統性能，更提高了系統的可擴展性。與此相對應，其他的郵件系統將盡可能多的信息都保存在中央數據庫中，隨著用戶數目的增大，效率會越來越低，最終成為系統的瓶頸。
5．HTTP session模塊 由于HTTP協議的無狀態性質，所以當用戶登陸后需要保存用戶的session(會話)信息。傳統的session模塊都是通過數據庫完成的，這樣，如果用戶量非常大，會對數據庫造成很大的負載，最終形成整套系統的瓶頸。ComConnect系統對session的處理，采用了自行開發的專門的session server處理，系統拿出一臺單獨的服務器充當session server，內部定義了一套session協議來維護每個session的狀態，并在一定時間客戶端沒有訪問時自動刪除session記錄以實現session垃圾回收機制。該模塊的數據庫采用了高效的內存數據庫：BerkeleyDB3。在這種設計下，web服務器作為session模塊的客戶端，session server作為session模塊的服務器端。當用戶登陸時，Web服務器會通過session協議訪問后臺的session server，以記錄該次session的信息。 與傳統的session管理機制相比，這種方式減少了中央數據庫的負載，又由于內存數據庫的高效性，以及session協議的簡單性，大大提高系統的響應速度。 6．認證模塊 系統需要認證的部分有：Web登陸、發信時的用戶身份認證、POP3登陸。這些認證都通過本地的認證模塊實現，該模塊工作機制與“數據庫中間件”類似，通過unix domain socket進行進程間通訊，并與數據庫持續連接，以及維護本地的緩存。 7．Web模塊 這是用戶使用的Web界面部分。該模塊通過IMAP協議，與后臺的“郵件存儲模塊”進行通訊。其中的session，是通過“HTTP session模塊”完成的。
8．郵件存儲模塊
該部分是整個系統最為核心的模塊，所有用戶的郵件最終都保存在該模塊上。每個用戶的郵箱就是一個目錄，每封郵件就是一個文件。
有些大規模郵件系統將郵件的信頭(header)保存到數據庫中，而郵件的正文(body)保存到文件中。這種設計，可以提高用戶在訪問郵箱時的速度，尤其是郵箱中有很多封郵件的時候，另外，在實際實現的時候也有代碼簡單的優點。但它的缺點也是突出的，首先，“收信模塊”要對進入的每封郵件進行處理，以提取出信頭和正文，這會降低收信的效率；另一方面，由于每封郵件都會引發對數據庫的INSERT操作，因此會加大數據庫的負擔；還有，隨著系統接收郵件數目的增大，數據庫中的記錄數也會相應地增大，最終可能出現瓶頸。
與上面的解決辦法不同，ComConnect系統在文件系統中建立了索引機制，而不采用數據庫的索引。每個用戶的郵箱(對應一個目錄)下，都有該用戶所有郵件的索引。這樣，當用戶操作郵箱時，速度可以非?？?；另外，對每封進入的郵件，“收信模塊”不需要做任何處理，可以直接通過LMTP通道傳輸給“郵件存儲模塊”，而“郵件存儲模塊”直接將內容寫入對應用戶的郵箱目錄下。
這八大模塊構成了ComConnect整個系統，其中每個模塊都具有很好的擴展機制，可以通過增加計算機數目來提高性能。更詳細的信息，請訪問： http://bms.elong.com/ads/
下面是數據庫中間件的部分代碼，讀者可以從中了解其運行機制：
dbServer dbserver = db_unknown;
int midInited = 0;
/*
*-------------------------------------------------------------- * * mid_init * *-------------------------------------------------------------- */ void mid_init(const char *conf) { const char fname[] = "mid_init"; const char *server ;
if ( !conf || !*conf ) {
myconfig_read(MIDWARE_CONFIG_FIENAME); } else { myconfig_read(conf); } server = myconfig_getstring("backend_server",NULL); if ( !server || !*server ) { syslog(LOG_ERR,"[%s]invalid 'backend_server' in conf file",fname); exit(1); /* yes, we exit directly */ }
/* judge what backend server is */
if ( !strcasecmp(server,BACKEND_MYSQL) ) { syslog(LOG_DEBUG3,"[%s]backend server is %s",fname,BACKEND_MYSQL); dbserver = db_mysql; } else if ( !strcasecmp(server,BACKEND_ORACLE) ) { syslog(LOG_DEBUG3,"[%s]backend server is %s",fname,BACKEND_ORACLE); dbserver = db_oracle; } else if ( !strcasecmp(server,BACKEND_BERKELEYDB) ) { syslog(LOG_DEBUG3,"[%s]backend server is %s",fname,BACKEND_BERKELEYDB); dbserver = db_berkeleydb; } else { dbserver = db_unknown; syslog(LOG_CRIT,"[%s]unknown backend server: %s",fname,server); exit(1); }
midInited = 1;
} /* mid_init */
/*
*------------------------------------------------------------------------ * * sendCmd * * -- midware protocol implementation on midwared client side * * ask midware server something via unix domain socket * * RET: * CMD_ERR_SYS: system error * CMD_INVALID: null cmd sent to midwared server * CMD_OK : ok * *------------------------------------------------------------------------ */ static cmdResult sendCmd(char *cmd,char *sqlcmd,const char *param2,const char *param3, const char *param4,char **reply) { const char fname[]="sendCmd"; int s; int r,iovcount=0; struct sockaddr_un srvaddr; struct iovec iov[6]; static char response[MAX_REP_LEN]; char sockfile[1024]; int start, n; if (reply) *reply = NULL; if ( !cmd || !*cmd ) { syslog(LOG_ERR,"[%s]null cmd specified",fname); return CMD_INVALID; } /* create socket and connect to midware server */ s = socket(AF_UNIX, SOCK_STREAM, 0); if (s == -1) { syslog(LOG_CRIT,"[%s]create socket failed: %m",fname); return CMD_ERR_SYS; } memset(sockfile,0,sizeof(sockfile)); strncpy(sockfile, myconfig_getstring("unixsock_dir",DEFAULT_MIDWARED_DIR) , sizeof(sockfile)); strcat(sockfile,"/"); strcat(sockfile,SOCKET_FILENAME); syslog(LOG_DEBUG3,"[%s]socket filename: %s",fname,sockfile); memset((char *)&srvaddr, 0, sizeof(srvaddr)); srvaddr.sun_family = AF_UNIX; strncpy(srvaddr.sun_path, sockfile, sizeof(srvaddr.sun_path)); r = connect(s, (struct sockaddr *)&srvaddr, sizeof(srvaddr)); if (r == -1) { if (reply) *reply="Cannot connect to midwared server"; syslog(LOG_ERR,"[%s]:connect: %m",fname); return CMD_ERR_SYS; } /* * connected with the unix-domain socket ,next * prepare the parameters for midwared server */ iov[iovcount].iov_base = (char *)cmd; iov[iovcount].iov_len = strlen(cmd)+1; /* check sqlcmd */ if ( sqlcmd && *sqlcmd ){ iovcount++; iov[iovcount].iov_base = (char *)sqlcmd; iov[iovcount].iov_len = strlen(sqlcmd)+1; } else { goto startsend; } /* check parameter2 */ if ( param2 && *param2 ){ iovcount++; iov[iovcount].iov_base = (char *)param2; iov[iovcount].iov_len = strlen(param2)+1; } else { goto startsend; } /* check parameter3 */ if ( param3 && *param3 ){ iovcount++; iov[iovcount].iov_base = (char *)param3; iov[iovcount].iov_len = strlen(param3)+1; } else { goto startsend; } /* check parameter4 */ if ( param4 && *param4 ){ iovcount++; iov[iovcount].iov_base = (char *)param4; iov[iovcount].iov_len = strlen(param4)+1; }
startsend:
retry_writev(s, iov, iovcount+1); /* get reply from midwared server */ start = 0; while (start n = read(s, response+start, sizeof(response) - 1 - start); if (n <1) break; start += n; } /* while */ close(s); /* marshell the reply for client call */ if ( start <) =1 { /* failurely got the reply */ if ( reply ) { *reply=response; } return CMD_ERR_SYS; } else { /* successfully got the reply */ response[start] = '\0'; if (reply) { *reply=response; } return CMD_OK; }
} /* sendCmd */
/*
*-------------------------------------------------------------- * * mid_query * - exported for end user directly * * ARG - * key: used for cache * * PRECON: * argument 'result' must be initialized before this call * *-------------------------------------------------------------- */ midQueryResult mid_query(char *sqlcmd,const char *key,int useCache,char result[]) { extern char *cache_mapstr(int); const char fname[]="mid_query"; char *reply; int r = 0;
if ( midInited == 0 ) {
mid_init(NULL); } switch ( dbserver ) { case db_mysql: r = sendCmd(REQ_MYSQL_QUERY,sqlcmd,key,cache_mapstr(useCache), NULL,(char **)&reply); break; case db_oracle: r = sendCmd(REQ_ORACLEL_QUERY,sqlcmd,key,cache_mapstr(useCache), NULL,(char **)&reply); break; case db_berkeleydb: break; default: syslog(LOG_CRIT,"[%s]unknown backend server",fname); break; } /* switch */
/* let midware client know what happened */
if ( result && *result ) strcpy(result,reply); if ( r != CMD_OK ) { syslog(LOG_DEBUG3,"[%s]no reply,system error",fname); return QUERY_SYS_ERR; }
if ( !strcmp(reply,REP_DB_QUERY_FAIL) ) {
return QUERY_FAIL; } else if ( !strcmp(reply,REP_DB_QUERY_NOTFOUND) ) { return QUERY_NOTFOUND; } else if ( !strcmp(reply,REP_DB_QUERY_MANYRECORD) ) { return QUERY_FOUNDMANY; } else { return QUERY_OK; }
} /* mid_query */
/*
*-------------------------------------------------------------- * * mid_insert * *-------------------------------------------------------------- */ midInsertResult mid_insert(char *sqlcmd) { const char fname[]="mid_insert"; char *reply; int r = 0; if ( midInited == 0 ) { mid_init(NULL); } switch ( dbserver ) { case db_mysql: r = sendCmd(REQ_MYSQL_INSERT,sqlcmd,NULL,NULL,NULL,(char **)&reply); break; case db_oracle: r = sendCmd(REQ_ORACLE_INSERT,sqlcmd,NULL,NULL,NULL, (char **)&reply); break; case db_berkeleydb: break; default: syslog(LOG_CRIT,"[%s]unknown backend server",fname); break; } /* switch */ if ( r != CMD_OK ) { syslog(LOG_DEBUG3,"[%s]no reply,system error",fname); return INSERT_SYS_ERR; }
if ( !strcmp(reply,REP_DB_INSERT_FAIL) ) {
return INSERT_FAIL; } else { return INSERT_OK; }
} /* mid_insert */
|