|
一、问题的提出 作者最近在开发基于Internet网上的可视电话过程中，碰到了这样一个问题。在基于Internet网上的可视电话系统中，同时要进行语音采集、语音编解码、图象采集、图象编解码、语音和图象码流的传输，所有的这些事情，都要并行处理。特别是语音信号，如果进行图象编解码时间过长，语音信号得不到服务，通话就有间断，如果图象或语音处理时间过长，而不能及时的传输码流数据，通信同样也会中断。这样就要求我们实现一种并行编程，在只有一个CPU的机器上，也就是要将该CPU时间按照一定的优先准则分配给各个事件，定期处理某一事件而不会在某一事件处理过长，在32位Windows95或WindowsNT下，我们可以用多线程的编程技术来实现这种并行编程。实际上，这种并行编程在很多场合下都是必须的。例如，在FileManager拷贝文件时，它显示一个对话框，列出源文件和目标文件的名称，并在对话框中包含了一个Cancel按钮。如果在文件拷贝过程中，点中Cancel按钮，就会终止拷贝。在16位Windows中，实现这类功能需要在FileCopy循环内部周期性地调用PeekMessage函数。如果正在读一个很大的数据块，则只有当这个块读完以后才能响应这个按钮动作，如果从软盘读文件，则要花费好几秒的时间，由于机器反应太迟钝，你会频繁地点中这个按钮，以为系统不知道你想终止这个操作。如果把FileCopy指令放入另外一个线程，你就不需要在代码中放一大堆PeekMessage函数，处理用户界面的线程将与它分开操作，这样，点中Cancel按钮后会立即得到响应。同样的道理，在应用程序中创建一个单独线程来处理所有打印任务也是很有用的，这样，用户可以在打印处理时继续使用应用程序。 二、线程的概念 为了了解线程的概念，我们必须先讨论一下进程的概念。 一个进程通常定义为程序的一个实例。在Win32中，进程占据4GB的地址空间。与它们在MS-DOS和16位Windows操作系统中不同，Win32进程是没有活力的。这就是说，一个Win32进程并不执行什么指令，它只是占据着4GB的地址空间，此空间中有应用程序EXE文件的代码和数据。EXE需要的任意DLL也将它们的代码和数据装入到进程的地址空间。除了地址空间，进程还占有某些资源，比如文件、动态内存分配和线程。当进程终止时，在它生命期中创建的各种资源将被清除。 但是进程是没有活力的，它只是一个静态的概念。为了让进程完成一些工作，进程必须至少占有一个线程，所以线程是描述进程内的执行，正是线程负责执行包含在进程的地址空间中的代码。实际上，单个进程可以包含几个线程，它们可以同时执行进程的地址空间中的代码。为了做到这一点，每个线程有自己的一组CPU寄存器和堆栈。每个进程至少有一个线程在执行其地址空间中的代码，如果没有线程执行进程地址空间中的代码，进程也就没有继续存在的理由，系统将自动清除进程及其地址空间。为了运行所有这些线程，操作系统为每个独立线程安排一些CPU时间，操作系统以轮转方式向线程提供时间片，这就给人一种假象，好象这些线程都在同时运行。创建一个Win32进程时，它的第一个线程称为主线程，它由系统自动生成，然后可由这个主线程生成额外的线程，这些线程，又可生成更多的线程。 三、线程的编程技术 1、编写线程函数 所有线程必须从一个指定的函数开始执行，该函数称为线程函数，它必须具有下列原型：DWORDWINAPIYourThreadFunc(LPVOIDlpvThreadParm); 该函数输入一个LPVOID型的参数，可以是一个DWORD型的整数，也可以是一个指向一个缓冲区的指针，返回一个DWORD型的值。象WinMain函数一样，这个函数并不由操作系统调用，操作系统调用包含在KERNEL32.DLL中的非C运行时的一个内部函数，如StartOfThread，然后由StartOfThread函数建立起一个异常处理框架后，调用我们的函数。 2、创建一个线程 一个进程的主线程是由操作系统自动生成，如果你要让一个主线程创建额外的线程，你可以调用来CreateThread完成。 HANDLECreateThread(LPSECURITY_ATTRIBUTESlpsa,DWORDcbstack,LPTHREAD_START_ROUTINElpStartAddr,LPVOIDlpvThreadParm,DWORDfdwCreate,LPDWORDlpIDThread); 其中lpsa参数为一个指向SECURITY_ATTRIBUTES结构的指针。如果想让对象为缺省安全属性的话，可以传一个NULL，如果想让任一个子进程都可继承一个该线程对象句柄，必须指定一个SECURITY_ATTRIBUTES结构，其中bInheritHandle成员初始化为TRUE。参数cbstack表示线程为自己所用堆栈分配的地址空间大小，0表示采用系统缺省值。参数lpStartAddr用来表示新线程开始执行时代码所在函数的地址，即为线程函数。lpvThreadParm为传入线程函数的参数，fdwCreate参数指定控制线程创建的附加标志，可以取两种值。如果该参数为0，线程就会立即开始执行，如果该参数为CREATE_SUSPENDED，则系统产生线程后，初始化CPU，登记CONTEXT结构的成员，准备好执行该线程函数中的第一条指令，但并不马上执行，而是挂起该线程。最后一个参数lpIDThread是一个DWORD类型地址，返回赋给该新线程的ID值。 3、终止线程 如果某线程调用了ExitThread函数，就可以终止自己。 VOIDExitThread(UINTfuExitCode); 这个函数为调用该函数的线程设置了退出码fuExitCode后，就终止该线程。 调用TerminateThread函数亦可终止线程。 BOOLTerminateThread(HANDLEhThread,DWORDdwExitCode); 该函数用来结束由hThread参数指定的线程，并把dwExitCode设成该线程的退出码。当某个线程不在响应时，我们可以用其他线程调用该函数来终止这个不响应的线程。 4、设定线程的相对优先级 当一个线程被首次创建时，它的优先级等同于它所属进程的优先级。在单个进程内可以通过调用SetThreadPriority函数改变线程的相对优先级。一个线程的优先级是相对于其所属的进程的优先级而言的。 BOOLSetThreadPriority(HANDLEhThread,intnPriority); 其中参数hThread是指向待修改优先级线程的句柄，nPriority可以是以下的值： THREAD_PRIORITY_LOWEST,THREAD_PRIORITY_BELOW_NORMAL,THREAD_PRIORITY_NORMAL,THREAD_PRIORITY_ABOVE_NORMAL,THREAD_PRIORITY_HIGHEST 5、挂起及恢复线程 先前我提到过可以创建挂起状态的线程（通过传递CREATE_SUSPENDED标志给函数CreateThread来实现）。当你这样做时，系统创建指定线程的核心对象，创建线程的栈，在CONTEXT结构中初始化线程CPU注册成员。然而，线程对象被分配了一个初始挂起计数值1，这表明了系统将不再分配CPU去执行线程。要开始执行一个线程，另一个线程必须调用ResumeThread并传递给它调用CreateThread时返回的线程句柄。DWORDResumeThread(HANDLEhThread); 一个线程可以被挂起多次。如果一个线程被挂起3次，则该线程在它被分配CPU之前必须被恢复3次。除了在创建线程时使用CREATE_SUSPENDED标志，你还可以用SuspendThread函数挂起线程。 DWORDSuspendThread(HANDLEhThread); 四、多线程编程技术的应用 我在前面说过，为了实现基于TCP/IP下的可视电话，就必须“并行”地执行语音采集、语音编解码、图象采集、图象编解码以及码流数据的接收与发送。语音与图象的采集由硬件采集卡进行，我们的程序只需初始化该硬件采集卡，然后实时读取采集数据即可，但语音和图象数据的编解码以及码流数据的传输都必须由程序去协调执行，决不能在某一件事件上处理过长，必须让CPU轮流的为各个事件服务，Windows95下的线程正是满足这种要求的编程技术。 下面我给出了利用Windows95环境下的多线程编程技术实现的基于TCP/IP的可视电话的部分源码，其中包括主窗口过程函数，以及主叫端与被叫端的TCP/IP接收线程函数和语音编解码的线程函数。由于图象编解码的实时性比语音处理与传输模块的实时性的要求要低些，所以我以语音编解码为事件去查询图象数据，然后进行图象编解码，而没有为图象编解码去单独实现一个线程。 在主窗口初始化时，我用CREATE_SUSPENDED标志创建了两个线程hThreadG7231和hThreadTCPRev。一个用于语音编解码，它的线程函数为G723Proc，该线程不断查询本地有无编好码的语音和图象的码流，如有，则进行H.223打包，然后通过TCP的端口发送给对方。另外一个线程用于TCP/IP的接收，它的线程函数为AcceptThreadProcRev，该线程不断侦测TCP/IP端口有无对方传来的码流，如有，就接收码流，进行H.223解码后送入相应的缓冲区。该缓冲区的内容，由语音编解码线程G723Proc查询，并送入相应的解码器。由于使用了多线程的编程技术，使得操作系统定时去服务语音编解码模块和传输模块，从而保证了通信的不中断。 五、程序源码 //基于TCP/IP可视电话主窗口的窗口过程 LONG APIENTRY MainWndProc(HWND hWnd,UINT message,UINT wParam, LONG lParam){ static HANDLE hThreadG7231,hThreadTCPListen,hThreadTCPRev; DWORDThreadIDG7231,ThreadIDTCPListen,ThreadIDTCPRev; static THREADPACK tp; static THREADPACK tp1; unsigned char Buf[80]; CAPSTATUS capStatus; switch (message) { case WM_CREATE: Init_Wsock(hWnd); //初始化一些数据结构 Init_BS(2,&bs); vd_tx_pdu.V_S = 0;vd_tx_pdu.N_S = 0; vd_rx_pdu.V_R = 0;vd_tx_sdu.bytes = 0; if( dnldProg ( hWnd, "h324g723.exe") ) {//装入语音编解码的DSP核心 MessageBox(hWnd,"Load G.723.1 Kernel Error","Error",MB_OK); PostQuitMessage(0); } else MessageBox(hWnd,"Load G.723.1 Kernel OK!","Indication",MB_OK); //创建语音编解码的线程 parag7231.hWnd = hWnd; hThreadG7231=CreateThread (NULL, 0, (LPTHREAD_START_ROUTINE)G723Proc, (G7231DATA *)?g7231,CREATE_SUSPENDED, (LPDWORD)&ThreadIDG7231); if (!hThreadG7231){ wsprintf(Buf, "Error in creating G7231 thread: %d",GetLastError()); MessageBox (hWnd, Buf, "WM_CREATE", MB_OK);} //创建TCP/IP接收线程 tp1.hWnd = hWnd; hThreadTCPRev = CreateThread (NULL, 0, (LPTHREAD_START_ROUTINE)AcceptThreadProcRev, (G7231DATA *)&tp1,CREATE_SUSPENDED, (LPDWORD)&ThreadIDTCPRev); if (!hThreadTCPRev){ wsprintf(Buf, "Error in creating TCP Receive thread: %d",GetLastError()); MessageBox (hWnd, Buf, "WM_CREATE", MB_OK);} //开始侦听网络 SendMessage(hWnd,WM_COMMAND,IDM_LISTEN,NULL); break; case WM_VIDEO_ENCODE: //图象编码 if(needencode)EncodeFunction(hWnd); needencode = SendVideoToBuff(&vd_tx_sdu, buff); frameMode=TRUE; capPreview(capWnd,FALSE);capOverlay(capWnd,FALSE); capGrabFrameNoStop(capWnd); break; case WM_VIDEO_DECODE: //图象解码 Video_Decod_begin = 1; play_movie(); Video_Decod_begin = 0; break; case WM_COMMAND: switch(LOWORD(wParam)) { case IDM_CONNECT: // 响应对方的呼叫，接通可视电话 WskConnect( hWnd ); ResumeThread(hThreadTCPRev);//运行TCP/IP接收线程 ResumeThread(hThreadG7231);//运行语音编解码线程 BeginG7231Codec(); //初始化图象采集卡，并开始采集图象 frameMode = FALSE; capWnd= capCreateCaptureWindow((LPSTR)"Capture Window", WS_CHILD | WS_VISIBLE, 100, 100, 176,144 , (HWND) hWnd, (int) 0) ; capSetCallbackOnError(capWnd, (FARPROC)ErrorCallbackProc) ; capSetCallbackOnStatus(capWnd, (FARPROC)StatusCallbackProc) ; capSetCallbackOnFrame(capWnd, (FARPROC)FrameCallbackProc) ; capDriverConnect(capWnd, 0); CenterCaptureWindow(hWnd, capWnd); capDlgVideoSource(capWnd); capDlgVideoFormat(capWnd); capDlgVideoCompression(capWnd); capGetStatus(capWnd,&capStatus,sizeof(CAPSTATUS)); StartNewVideoChannel(hWnd, capWnd) ; image = image_one; frameMode = TRUE; capPreview(capWnd,FALSE); capOverlay(capWnd,FALSE); capGrabFrameNoStop(capWnd); break; case IDM_LISTEN: //拨对方号码，呼叫对方 sock = socket( AF_INET, SOCK_STREAM, 0); if (sock == INVALID_SOCKET) { MessageBox(hWnd, "socket() failed", "Error", MB_OK); closesocket(sock); break;} if (!FillAddr(hWnd, &local_sin, FALSE )) // 获取TCP/IP地址和端口号 break; EnableMenuItem(GetMenu( hWnd ), IDM_LISTEN, MF_GRAYED); SetWindowText( hWnd, "Waiting for connection.."); bind ( sock , (struct sockaddr FAR *)&local_sin,sizeof(local_sin); if (listen( sock, MAX_PENDING_CONNECTS ) <0) { sprintf(szBuff, "%d is the error", WSAGetLastError()); MessageBox(hWnd, szBuff, "listen(sock) failed", MB_OK); break;} tp.hWnd="hWnd;//开始本地的TCP/IP接收线程" _beginthread(AcceptThreadProc,0,&tp); ResumeThread(hThreadG7231); // 开始本地语音编解码的线程 break; case IDM_DISCONNECT: //挂断可视电话 CloseG7231Codec(); SuspendThread(hThreadG7231); SuspendThread(hThreadTCPRev); WSACleanup(); Init_Video_Decod_Again(); capSetCallbackOnError(capWnd, NULL) ; capSetCallbackOnStatus(capWnd, NULL) ; InvalidateRect(hWnd,NULL,1); capSetCallbackOnFrame(capWnd, NULL) ; capSetCallbackOnVideoStream(capWnd, NULL) ; capDriverDisconnect(capWnd); Init_Wsock(hWnd); MessageBox(hWnd, "Now closing the Video telephone","",MB_OK); SetDisConnectMenus(hWnd); SendMessage(hWnd, WM_COMMAND,IDM_LISTEN,NULL); break; case IDM_EXIT: CloseG7231Codec(); SendMessage(hWnd, WM_CLOSE, 0, 0l); break; default: return (DefWindowProc(hWnd, message, wParam, lParam)); } break; case WM_CLOSE: if (IDOK !="MessageBox(" hWnd, "OK to close window?", gszAppName, MB_ICONQUESTION | MB_OKCANCEL )) break ; case WM_DESTROY: WSACleanup(); CloseG7231Codec(); TerminateThread(hThreadG7231,0); TerminateThread(hThreadTCPRev,0); capSetCallbackOnError(capWnd, NULL) ; capSetCallbackOnStatus(capWnd, NULL) ; capSetCallbackOnFrame(capWnd, NULL) ; capSetCallbackOnVideoStream(capWnd, NULL) ; capDriverDisconnect(capWnd); FreeAll(); PostQuitMessage(0); break; default: /* Passes it on if unproccessed*/ return (DefWindowProc(hWnd, message, wParam, lParam)); } return (0); } //主叫方TCP/IP接收线程 DWORD WINAPI AcceptThreadProc( PTHREADPACK ptp ){ SOCKADDR_IN acc_sin;/* Accept socket address internet style*/ int acc_sin_len;/* Accept socket address length */ int status; acc_sin_len="sizeof(acc_sin);" //调用阻塞函数accept,一直到远端响应为止 sock="accept(" sock,(struct sockaddr FAR *) &acc_sin,(int FAR *) &acc_sin_len ); if (sock < 0) { sprintf(szBuff, "%d is the error", WSAGetLastError()); MessageBox(ptp->hWnd, szBuff, "accept(sock) failed", MB_OK); return (1); } SetConnectMenus( ptp->hWnd ); //远端提机，可视电话接通 BeginG7231Codec(); while (1) { beg1: status = recv((SOCKET)sock, r_mux_buf, MY_MSG_LENGTH, NO_FLAGS_SET ); if (status == SOCKET_ERROR) { status = WSAGetLastError(); if( status == 10054 ){ MessageBox(ptp->hWnd,"对方挂断电话","Indication", MB_OK); SendMessage( ptp->hWnd, WM_COMMAND,IDM_DISCONNECT,NULL); _endthread(); return (1); } goto beg1; } if (status) { r_mux_buf[ status ] = '\0'; if( r_mux_buf_filled == 1 ) r_mux_buf_overwrite = 1; else r_mux_buf_filled = 1; r_mux_buf_length = status; } else { MessageBox( hWnd, "Connection broken", "Error", MB_OK); SendMessage( ptp->hWnd, WM_COMMAND, IDM_DISCONNECT,NULL); _endthread(); return (2); } demux(); //线路码流H.223解码 } return (0); } //被叫方TCP/IP接收线程 DWORD WINAPI AcceptThreadProcRev( PTHREADPACK ptp ){ int status; while (1) { beg2:status = recv((SOCKET)sock, r_mux_buf, MY_MSG_LENGTH, NO_FLAGS_SET ); if (status == SOCKET_ERROR) { status =WSAGetLastError(); if( status == 10054 ){ MessageBox(ptp->hWnd,"对方挂断电话","Indication", MB_OK); SendMessage( ptp->hWnd, WM_COMMAND, IDM_DISCONNECT,NULL); return (1); } goto beg2; } if (status) { r_mux_buf[ status ] = '\0'; if( r_mux_buf_filled == 1 ) r_mux_buf_overwrite = 1; else r_mux_buf_filled = 1; r_mux_buf_length = status; } else { MessageBox( hWnd, "Connection broken", "Error", MB_OK); SendMessage( ptp->hWnd, WM_COMMAND, IDM_DISCONNECT,NULL); return (2); } demux(); }/* while (forever) */ return (0); } //语音编解码线程 DWORD WINAPI G723Proc(G7231DATA *data){ int i,len; Audio_tx_pduad_tx_pdu; unsigned char mux[MAX_MUX_PDU_SIZE]; do { len = 0; //检测本地有无语音，图象码流要传输 i = DetectAudioVideoData(); switch(i){ case AUDIO_ONLY: //只有语音码流 AL2_CRC_coder(&ad_tx_sdu,&ad_tx_pdu); //H.223打包 len = AL2_To_MUX(&ad_tx_pdu, mux); break; case VIDEO_ONLY://只有图象码流 SDU_To_PDU(&vd_tx_sdu,&vd_tx_pdu); tx_AL3_I_PDU(&vd_tx_pdu ,&bs , 1); //H.223打包 len = AL3_To_MUX(&vd_tx_pdu,mux); break; case AUDIO_VIDEO://语音和图象码流 AL2_CRC_coder(&ad_tx_sdu,&ad_tx_pdu); SDU_To_PDU(&vd_tx_sdu,&vd_tx_pdu); tx_AL3_I_PDU(&vd_tx_pdu ,&bs , 1); //H.223打包 len = AL2_AL3_To_MUX(&ad_tx_pdu,&vd_tx_pdu,mux); break; case NO_AUDIO_VIDEO://此刻无码流要传输 break; } //TCP/IP发送码流 if(len != 0) send((SOCKET)sock,mux,len,0); //是否接收到待解码的码流，有就调用解码器 PutVideoStreamToDecod(); } while(1); return (0); } | 