千萬級郵件系統(tǒng)的設計及數(shù)據(jù)庫中間件部分實現(xiàn)
(作者：藝龍公司技術研發(fā)經(jīng)理 高鵬)
近日，藝龍公司推出了“藝龍五星級企業(yè)郵箱系統(tǒng)”和“藝龍個人5秒郵”的收費郵件系統(tǒng)，這兩套系統(tǒng)都采用的是藝龍公司與全世界最大的郵件運營商mail.com合作設計和開發(fā)的千萬級郵件系統(tǒng)：“ComConnect”。筆者是該系統(tǒng)的主要設計人員，本文將介紹這套系統(tǒng)的技術特色，以及數(shù)據(jù)庫中間件部分實現(xiàn)。 藝龍公司的“ComConnect”系統(tǒng)可以運行在大部分的unix系統(tǒng)平臺，包括solaris、linux等，采用模塊化的設計，整個系統(tǒng)分為下面八大模塊： 1．數(shù)據(jù)庫中間件模塊＋緩存模塊 這是整套系統(tǒng)的主線，貫穿于系統(tǒng)的其他所有模塊。 在大規(guī)模郵件系統(tǒng)中，為了管理的方便和系統(tǒng)的分布式部署，都需要采用SQL數(shù)據(jù)庫保存一些信息，例如帳號信息等。例如，當用戶通過POP3登陸時，服務器端會啟動一個程序，與SQL數(shù)據(jù)庫連接，并查詢該用戶身份的合法性，當用戶退出時，該程序會斷開與SQL數(shù)據(jù)庫的連接。這樣，大量用戶的登陸、退出，都會對應于SQL數(shù)據(jù)庫的連接、斷開操作，降低了系統(tǒng)的效率，也增加了SQL數(shù)據(jù)庫的負載。
在ComConnect系統(tǒng)中，所有數(shù)據(jù)庫的操作，都是通過“數(shù)據(jù)庫中間件”模塊完成的。該模塊會保持與SQL數(shù)據(jù)庫的連接，永不斷開。當有程序需要查詢數(shù)據(jù)庫時，會通過unix domain socket與本地的數(shù)據(jù)庫中間件服務器相連接，中間件會取得查詢需求，并進行實際的SQL查詢，并將結果通過unix domain socket返回給應用程序。同時，對于所得的查詢，中間件服務器會進行本地緩存，保存到內(nèi)存數(shù)據(jù)庫BerkeleyDB3(http://www.sleepycat.com)中，這樣下次同樣的查詢請求，就不必SQL查詢了，因為結果已經(jīng)被保存在本地緩存了。這種設計，一方面，避免了頻繁的SQL數(shù)據(jù)庫的連接、斷開的操作，另一方面，由于本地緩存的存在，也大大降低了SQL數(shù)據(jù)庫的負載，提高了系統(tǒng)的響應效率。
2．收信模塊
該模塊對應于DNS中的MX記錄。
傳統(tǒng)的收信模塊運行方式是，在得到一封郵件后，保存到本地，或通過SMTP協(xié)議進行轉發(fā)。而在ComConnect系統(tǒng)中，采用LMTP(local mail transport protocol)協(xié)議連接“收信模塊”和“郵件存儲模塊”，在這兩個模塊之間，永久保持若干個通道，當收信模塊得到新的郵件時，就會利用這些已經(jīng)存在的通道將郵件內(nèi)容傳輸給后臺的郵件存儲模塊。這種設計，大大減少了SMTP服務器頻繁的fork、exit過程，從而提高了效率。 3．發(fā)信模塊 該模塊是整個系統(tǒng)中最獨立的模塊。為了提高安全性，ComConnect系統(tǒng)通過cyrus-sasl實現(xiàn)了發(fā)信服務器的用戶身份認證功能。認證部分通過“認證模塊”完成。 4．中央數(shù)據(jù)庫模塊
與其他大規(guī)模郵件系統(tǒng)一樣，ComConnect系統(tǒng)也需要中央數(shù)據(jù)庫(支持SQL標準)模塊。但與其他系統(tǒng)不同的是，ComConnect系統(tǒng)的中央數(shù)據(jù)庫保存的內(nèi)容非常少，僅僅保存用戶帳號信息。而且，由于“數(shù)據(jù)庫中間件”和“緩存”模塊的存在，在理想狀態(tài)下，中央數(shù)據(jù)庫根本不需要存在。
這種設計，極大地提高了系統(tǒng)性能，更提高了系統(tǒng)的可擴展性。與此相對應，其他的郵件系統(tǒng)將盡可能多的信息都保存在中央數(shù)據(jù)庫中，隨著用戶數(shù)目的增大，效率會越來越低，最終成為系統(tǒng)的瓶頸。
5．HTTP session模塊 由于HTTP協(xié)議的無狀態(tài)性質(zhì)，所以當用戶登陸后需要保存用戶的session(會話)信息。傳統(tǒng)的session模塊都是通過數(shù)據(jù)庫完成的，這樣，如果用戶量非常大，會對數(shù)據(jù)庫造成很大的負載，最終形成整套系統(tǒng)的瓶頸。ComConnect系統(tǒng)對session的處理，采用了自行開發(fā)的專門的session server處理，系統(tǒng)拿出一臺單獨的服務器充當session server，內(nèi)部定義了一套session協(xié)議來維護每個session的狀態(tài)，并在一定時間客戶端沒有訪問時自動刪除session記錄以實現(xiàn)session垃圾回收機制。該模塊的數(shù)據(jù)庫采用了高效的內(nèi)存數(shù)據(jù)庫：BerkeleyDB3。在這種設計下，web服務器作為session模塊的客戶端，session server作為session模塊的服務器端。當用戶登陸時，Web服務器會通過session協(xié)議訪問后臺的session server，以記錄該次session的信息。 與傳統(tǒng)的session管理機制相比，這種方式減少了中央數(shù)據(jù)庫的負載，又由于內(nèi)存數(shù)據(jù)庫的高效性，以及session協(xié)議的簡單性，大大提高系統(tǒng)的響應速度。 6．認證模塊 系統(tǒng)需要認證的部分有：Web登陸、發(fā)信時的用戶身份認證、POP3登陸。這些認證都通過本地的認證模塊實現(xiàn)，該模塊工作機制與“數(shù)據(jù)庫中間件”類似，通過unix domain socket進行進程間通訊，并與數(shù)據(jù)庫持續(xù)連接，以及維護本地的緩存。 7．Web模塊 這是用戶使用的Web界面部分。該模塊通過IMAP協(xié)議，與后臺的“郵件存儲模塊”進行通訊。其中的session，是通過“HTTP session模塊”完成的。
8．郵件存儲模塊
該部分是整個系統(tǒng)最為核心的模塊，所有用戶的郵件最終都保存在該模塊上。每個用戶的郵箱就是一個目錄，每封郵件就是一個文件。
有些大規(guī)模郵件系統(tǒng)將郵件的信頭(header)保存到數(shù)據(jù)庫中，而郵件的正文(body)保存到文件中。這種設計，可以提高用戶在訪問郵箱時的速度，尤其是郵箱中有很多封郵件的時候，另外，在實際實現(xiàn)的時候也有代碼簡單的優(yōu)點。但它的缺點也是突出的，首先，“收信模塊”要對進入的每封郵件進行處理，以提取出信頭和正文，這會降低收信的效率；另一方面，由于每封郵件都會引發(fā)對數(shù)據(jù)庫的INSERT操作，因此會加大數(shù)據(jù)庫的負擔；還有，隨著系統(tǒng)接收郵件數(shù)目的增大，數(shù)據(jù)庫中的記錄數(shù)也會相應地增大，最終可能出現(xiàn)瓶頸。
與上面的解決辦法不同，ComConnect系統(tǒng)在文件系統(tǒng)中建立了索引機制，而不采用數(shù)據(jù)庫的索引。每個用戶的郵箱(對應一個目錄)下，都有該用戶所有郵件的索引。這樣，當用戶操作郵箱時，速度可以非常快；另外，對每封進入的郵件，“收信模塊”不需要做任何處理，可以直接通過LMTP通道傳輸給“郵件存儲模塊”，而“郵件存儲模塊”直接將內(nèi)容寫入對應用戶的郵箱目錄下。
這八大模塊構成了ComConnect整個系統(tǒng)，其中每個模塊都具有很好的擴展機制，可以通過增加計算機數(shù)目來提高性能。更詳細的信息，請訪問： http://bms.elong.com/ads/
下面是數(shù)據(jù)庫中間件的部分代碼，讀者可以從中了解其運行機制：
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 */
|