程式師世界 >> 編程語言 >> C語言 >> VC >> 關於VC++ >> 基於select I/O模型的遠程目錄浏覽與多線程文件下載

基於select I/O模型的遠程目錄浏覽與多線程文件下載

編輯：關於VC++

摘要：

服務器端基於select I/O模型。為防止程序界面阻塞, 有一個子線程用於不斷接收socket並select其中的處理。客戶端只有一個線程函數, 不過其利用率很高, 可用於遠程目錄交換, 請求文件大小, 創建若干線程來下載文件。文件傳輸有上傳和下載,還有對等傳輸, 這個項目中, 傳輸文件具體指下載。

正文：

一　數據及數據結構

1 傳送包, 客戶端主線程給子線程傳遞的結構體

typedef struct{ 　　　　char packetType;　// 請求類型 r：request rootdriver　d：directory f：file D：:data 　　　　unsigned int length;　//用於傳送int類型　　　　char content[2000];　//傳送的內容　　　　}DATA_PACKET;

請求類型：'r'：服務器邏輯盤符, d; 盤符和文件夾下的文件夾和文件名, f'：文件大小, D：文件內容。

在客戶端請求下載文件, length末兩位用於記錄下載的線程總數, 前面的用於記錄當前的線程序號。在服務器端這個反運算很容易實現。

在請求文件大小和請求下載時, content都用於保存請求文件的絕對路徑.這個在服務器端做反運算要用自定義函數處理。

2　全局變量(用於線程間的通信)

char* pDrives;　// 數據緩沖 CString arrFiles[200];　//文件目錄字符串數組 CString savePath;　//文件保存路徑 long fSize=0l, ,recvFSize=0l;　//文件大小,已經接收的文件大小 CString strIP;　//IP bool thrFlag=false;　//下載線程創建置位

理論上講, 在程序中應盡量避免使用全局變量,因為破壞程序結構, 君不見Java和C#完全面向類。但為了方便, 所以就用了全局變量, 且是應用程序級的, 這樣線程間的通訊很方便。值得一提的是, 這裡thrFlag很重要, 它涉及到確保線程創建一定正確的問題, 在後面“要注意的幾點問題”中會更詳細的討論。

3 目錄樹類CExtnTreeCtrl::CTreeCtrl

該類繼承於CTreeCtrl類, 主要擴展獲得樹的某個項目在整棵樹的完整路徑, 以便把這個路徑規格化後能向服務器請求文件。在這裡還實現了將一個完整路徑轉為文件名的函數。

對於如何構造這棵樹, 有一個很方便高效的辦法：雙擊樹,把雙擊的項作為根.如果為盤符或文件夾, 就發送請求給服務器, 待服務器返回, 自動填充在這個項下。如果為文件, 則該項名有擴展名, 將請求下載。這個辦法操作起來方便, 而且還提高了系統性能, 至少在局域網是這樣。如果就一次從服務器中請求整個文件系統的目錄內容, 肯定會慢很多。

4.參數設置對話框類 CSetParam

這個類用於設置下載的線程數目, 和默認的保存文件夾路徑。這個類會寫一個“Setting.ini”文件在C盤以保存參數。具體來說,如果用戶一直都沒有設置這些參數的話, 那麼GetPrivateProfileString(...)試圖讀取"C:\Setting.ini"文件會返回默認的線程數0, 和一個不是表示路徑的字條串“defaultpath”, 這時, 下載程序會自動設置參數, 分別為3, “C:\”。

二　幾個要注意的問題

1　MFC與Windows API

就多線程編寫網絡程序而言, 如果使用MFC的CAsyncSocket或CSocket, 主線程給子線程傳參量是一個非常頭痛的問題, 如果用Windows API實現就靈活多了。在文件操作方面, CFile還不錯。盡管如此, 我還是用了Windows API函數, 為了設計滿足要求。

2　socket傳送字符串或字符串數組

理論上, send是底層函數, 只要指定緩沖區首地址指針和緩沖區大小, 不管什麼狗屎垃圾它都會幫你把這片內存的內容send 出去。然而, 如果是字符串數組甚至字符串, 或者包含這兩者其中之一的封包, 內容是被 send 走了, 接收端緩沖區也顯示接收到了, 不過不是你想要的內容, 再看字符串(數組)還是空空然。

山重水復, 得找出路才有柳暗花明。轉轉思維, 用另一種做法吧。在Windows中, 文件名是不能含有 "|,<,>,%...." 等特殊字符的.由於恰好要傳的內容為目錄字符串數組, 所以把字條串數組轉為字符數組, 每個字符串用一個特殊字符隔開就OK了。

3. 確保連續創建線程正確性

用循環語句創建線程, 如果沒有一定的保護機制, 肯定會出事。程序如：

int i = paramSettingDlg．thrTotal　　　 //abtain download threads total 　　for( int j=1;j<=i;j++ ) 　　{ 　　　　dataPacket.length=j*100+i;　// download thread information. 　　::CreateThread(NULL, 0, ThreadDownload, (LPVOID)&dataPacket, 0, 0); 　　}

i為下載線程的總數, j為當前線程序號．主線程循環不管CreateThread創建成功與否還繼續執行, 當真正CreateThread在創建線程時, j的值可能已經被修改了．程序修改如下：

for( int j=1;j<=i;j++ ) 　　{ 　　　　dataPacket.length=j*100+i;　// download thread information. 　　　while( ::CreateThread(NULL, 0, ThreadDownload, (LPVOID)&dataPacket, 0, 0) ==0 ) 　　 Sleep(30); 　　}

在本項目中, 用的是置位機制.線程創建成功, 由線程置位為1, 主線程睡眼等待, 檢測到位為1時, 則置位為0並繼續創建線程．

4　打開關閉文件的控制

多線程讀/寫文件是一件很混亂的事, 有點像一堆乞丐搶飯吃那樣．所以要有條理就要維護秩序, 要維護秩序, 自然要犧牲性能．每次讀/寫文件都要定位到該包內容對應原文件的位置.這個應用程序是鎖定讀/寫一個包的時間．如果鎖定整個線程, 即等一個線程讀/寫該線程應該讀/寫的部分再解鎖, 系統性能必會急劇下降, 具體操作可以看"程序流程圖"。

另外, 線程寫入的數據要及時刷新文件流。否則, 程序等到緩沖區滿時才真正寫入文件的, 結果就是文件亂七八糟了。

5　雙循環程序

這個問題是在這個項目編程遇到的.雙循環的程序如下：

CString strArray[6]; 　　char* pChar = "abc1|abc2|abc3|ac"; 　　int len = strlen( "abc1|abc2|abc3|ac" ); 　　int i=0,j=0; 　　while( i<= len && j<=6 ) 　　{ 　　　　int k = 0; 　　　　while( pChar[i]!='|' ) 　　　　{ 　　　　　　strArray[j].Insert( k, pChar[i] ); 　　　　　　i++; 　　　　　　k++; 　　　　} 　　　　i++; 　　　　j++; 　　}

調試的時候, i>len程序像得了瘋牛病一樣還是往前跑, 不會停下來。要確保正確性, 內循環應該修改如：

while( pChar[i]!='|' && i<=len )

6　 memcpy預分配空間

在程序中對全局變量已經分配了一次空間char[2000], 在後續的使用中程序不會有問題, 但當整個程序關閉時卻冒出一個提示框說程序遇到一個問題需要關閉。把後續使用memcpy對這個變量賦值時, 如果小於1993個char空間就不會出現前面說到的提示框。

後來試著在第二次使用時再次給該變量分配了一次空間, 之後一切正常了。不知何解, 睇來memcpy的確不是安全的東西。

三　重要程序說明：

1　select I/O模型框架

select I/O模型對於大訪問量的網絡特別有效。在服務器端, 可以建立多個線程, 每個線程創建可同時創建一個讀寫fd_set。 fd_set的默認大小為64, 即在默認的情況下一個Set最多可以接受64個socket的連接。

DWORD WINAPI CFileTransSvrView::ThreadSelect(LPVOID lpParameter) { 　　// 初始化fd_set 　　SOCKET sListen=(SOCKET)lpParameter; 　　fd_set fdSocket; 　　FD_ZERO(&fdSocket); 　　FD_SET(sListen,&fdSocket); 　　// 不斷循環遍歷fd_set, 如果某項置位, 則表示該socket可用, 如果狀態為正在偵聽, 　　// 則建立連接, 並把它加入到fd_set中, 否則就等待接收數據。select有自動機制把不可　　//用的socket從fd_set中刪除。　　while(TRUE) 　　{ 　　　　fd_set fdRead=fdSocket; 　　　　int nRet＝::select(0, &fdRead, NULL, NULL, NULL); 　　　　if(nRet>0) 　　　　{ 　　　　　　for(int i＝0;i<(int)fdSocket.fd_count;i++) 　　　　　　{ 　　　　　　　　if(FD_ISSET(fdSocket.fd_array[i],&fdRead)) 　　　　　　　　{ 　　　　　　　　　　if(fdSocket.fd_array[i] == sListen) 　　　　　　　　　　{ 　　　　　　　　　　　　sockaddr_in addrRemote; 　　　　　　　　　　　　int nAddrLen=sizeof(addrRemote); 　　　　　　　　　　　　SOCKET sNew＝::accept(sListen,(sockaddr*)&addrRemote,&nAddrLen); 　　　　　　　　　　　　FD_SET(sNew,&fdSocket); 　　　　　　　　　　} 　　　　　　　　　　else 　　　　　　　　　　{ 　　　　　　　　　　　　DATA_PACKET recvPacket; 　　　　　　　　　　　　int nRecev＝::recv(fdSocket.fd_array[i], (char*)&recvPacket, 　　　　　　　　　　　　　　sizeof(recvPacket), 0)。　　　　　　　　　　　　if(nRecev>0) 　　　　　　　　　　　　{ 　　　　　　　　　　　　　　// respond request 　　　　　　　　　　　　} 　　　　　　　　　　　　else 　　　　　　　　　　　　{ 　　　　　　　　　　　　　　::closesocket(fdSocket.fd_array[i]); 　　　　　　　　　　　　　　FD_CLR(fdSocket.fd_array[i], &fdSocket); 　　　　　　　　　　　　} 　　　　　　　　　　} 　　　　　　　　} 　　　　　　} 　　　　} 　　} 　　return 0; }

2　文件下載函數

void CFileTransCltView::FillInFile(SOCKET socket, int thrInfo, int packIndex, char* pPackCont) { 　　// thrInfo末兩位為下載線程總數,再上兩位為當前線程序號　　int thrIndex = thrInfo/100;　// 當前線程序號　　int thrTotal = thrInfo-thrIndex*100;　// 下載線程的總數　　int thrLength = fSize/thrTotal;　// 當前線程要下載的文件長度　　int bPoint = (thrIndex-1)*thrLength;　// 當前線程要下載的文件起點　　if( thrIndex == thrTotal ) 　　　　thrLength = fSize-bPoint;　// 當前線程等於線程總數,則重新賦予要下載的文 //件長度　　FILE *file; 　　if( (file = fopen( savePath, "ab" )) == NULL ) 　　　　AfxMessageBox( "Open file occur an error\n" ); 　　// 每個線程都會試圖打開文件, 如果已打開, 返回文件流指針。為什麼這樣？因為　　//在TCP下載文件時, 線程的包是按順序來的, 但線程卻不一定會按前後順序執行。　　int packTotal = thrLength/2000 +1; 　　int subTotal = packTotal; 　　DATA_PACKET dPacket; 　　while( subTotal>0 ) 　　{ 　　　　critSection.Lock(5000); 　　　　fseek( file,bPoint+(packIndex-1)*2000,SEEK_SET ); 　　　　if( packIndex == packTotal ) //the last packet in thread,data maybe less than 2000B 　　　　{ 　　　　　　int fLength = thrLength-(packTotal-1)*2000; 　　　　　　int errorcode=fwrite( pPackCont,1,fLength,file); 　　　　　　fflush(file); 　　　　　　recvFSize+=fLength; 　　　　} 　　　　else 　　　　{ 　　　　　　fwrite( pPackCont,sizeof(char),2000,file ); 　　　　　　fflush(file); 　　　　　　recvFSize+=2000; 　　　　} 　　　　critSection.Unlock(); 　　　　subTotal－－; 　　　　if( subTotal>0 ) 　　　　{ 　　　　　　if( recv( socket, (char*)&dPacket, sizeof(dPacket), 0 ) != 0 ) 　　　　　　{ 　　　　　　　　packIndex = dPacket.length; 　　　　　　　　Memcpy( pPackCont, dPacket.content, sizeof(dPacket.content) ); 　　　　　　} 　　　　} 　　　　else 　　　　{ // 該線程下載任務完成, 判斷是否已經完全下載完文件, 如果是則關閉　　　　　　if( recvFSize == fSize ) 　　　　　　　　fclose(file); 　　　　　　return; 　　　　} 　　} }

3. 客戶端線程函數

DWORD WINAPI CFileTransCltView::ThreadRequest(LPVOID lpParameter) { 　　DATA_PACKET requestPacket=*(DATA_PACKET*)lpParameter; 　　SOCKET socket=::socket(AF_INET,SOCK_STREAM,IPPROTO_TCP); 　　sockaddr_in sin; 　　sin.sin_addr.S_un.S_addr=inet_addr(strIP); 　　sin.sin_family=AF_INET; 　　sin.sin_port=htons(1032); 　　if(::connect(socket,(sockaddr*)&sin,sizeof(sin))!=0) 　　{ 　　　　AfxMessageBox("Request failure!"); 　　　　return 0; 　　} 　　int thrInfo; 　　if( requestPacket.packetType=='D' ) //save download thead infomation for function 　　　　// FillInFile() 　　{ 　　　　thrInfo=requestPacket.length; 　　　　thrFlag=true; 　　} 　　::send( socket,(char*)&requestPacket,sizeof(requestPacket),0 ); 　　memset( requestPacket.content,0,sizeof(requestPacket.content) ); 　　if( ::recv(socket,(char*)&requestPacket,sizeof(requestPacket),0) ) 　　{ 　　　　if( requestPacket.packetType=='r' ) 　　　　{ 　　　　　　int szBuffer=requestPacket.length; 　　　　　　pDrives=new char[2000];　//Make sure that allocate memery! 　　　　　　memcpy(pDrives,requestPacket.content,szBuffer);　//Make sure that use memcpy() 　　　　} 　　　　if(requestPacket.packetType=='d' ) 　　　　{ 　　　　　　pDrives=new char[2000];　// 再次為pDrives分配空間才不會出錯　　　　　　Memcpy( pDrives,requestPacket.content,sizeof(requestPacket.content) ); 　　　　　　/* 2000 will take an error ?????*/ 　　　　　　//char's pointer to CString Array 　　　　　　int i=0,j=0; 　　　　　　while( i<requestPacket.length ) 　　　　　　{ 　　　　　　　　for(int k=0; pDrives[i]!='|' && i<requestPacket.length ; k++,i++ ) 　　　　　　　　arrFiles[j].Insert(k,pDrives[i]); 　　　　　　　　i++; 　　　　　　　　j++; 　　　　　　} 　　　　　　return 0; 　　　　} 　　　　if( requestPacket.packetType=='f' ) 　　　　　　fSize=requestPacket.length; 　　　　//Download file 　　　　if( requestPacket.packetType=='D' ) 　　　　{ 　　　　　　//memset( pDrives,0,2000 ); 　　　　　　Memcpy( pDrives,requestPacket.content,sizeof(requestPacket.content) ); 　　　　　　FillInFile( socket,thrInfo,requestPacket.length,pDrives); 　　　　} 　　} 　　::closesocket(socket); 　　return 0; }