socket 多线程写

现在项目有个场景：从一个socket来的数据 需要分发到多个逻辑线程处理；

多个逻辑线程需要通过同一个socket发送数据，是需要封装一个同步写

比如:

int SyncWrite(char* pData, int length){

CBuffer* pBuffer = new CBuffer(pData, length);

Lock clock;

socket.send(pBuffer.data, pBuffer.length);

}

------解决思路----------------------

一个socket只让一个线程读和写。不应该多个线程都能对这个socket进行读或者写。

理由很简单，就像楼主的代码，如果多个线程都要操纵socket，然后用一个锁进行同步。假如send是阻塞的，多个线程同时想往socket写入数据，那么只有其中一个能调用send，其他线程只能被锁住，效率低。如果send是非阻塞的，那么可能发生A线程send一半数据就返回了，然后B线程接着send数据。这样会把数据搞乱的。

应该用一个线程专门负责socket的读和写。其他线程想写数据就把数据放到一个队列里面，然后那个负责socket读写的线程就去读这个队列，读到数据后就send。

------解决思路----------------------

搞个收发缓冲队列。

------解决思路----------------------

用写缓存队列，逻辑线程将要写入socket的数据扔入队列，由专门的写socket线程从队列中取出并进行socket写入。队列可以采用信号量进行事件通知。

http://blog.csdn.net/truexf/article/details/38342983

------解决思路----------------------

仅供参考//循环向a函数每次发送200个字节长度(这个是固定的)的buffer,

//a函数中需要将循环传进来的buffer，组成240字节(也是固定的)的新buffer进行处理，

//在处理的时候每次从新buffer中取两个字节打印

#ifdef WIN32

#pragma warning(disable:4996)

#endif

#include 

#include 

#include 

#ifdef WIN32

#include 

#include 

#include 

#define  MYVOID             void

#define  vsnprintf          _vsnprintf

#else

#include 

#include 

#include 

#define  CRITICAL_SECTION   pthread_mutex_t

#define  MYVOID             void *

#endif

//Log{

#define MAXLOGSIZE 20000000

#define MAXLINSIZE 16000

#include 

#include 

#include 

char logfilename1[]="MyLog1.log";

char logfilename2[]="MyLog2.log";

static char logstr[MAXLINSIZE+1];

char datestr[16];

char timestr[16];

char mss[4];

CRITICAL_SECTION cs_log;

FILE *flog;

#ifdef WIN32

void Lock(CRITICAL_SECTION *l) {

EnterCriticalSection(l);

}

void Unlock(CRITICAL_SECTION *l) {

LeaveCriticalSection(l);

}

void sleep_ms(int ms) {

Sleep(ms);

}

#else

void Lock(CRITICAL_SECTION *l) {

pthread_mutex_lock(l);

}

void Unlock(CRITICAL_SECTION *l) {

pthread_mutex_unlock(l);

}

void sleep_ms(int ms) {

usleep(ms*1000);

}

#endif

void LogV(const char *pszFmt,va_list argp) {

struct tm *now;

struct timeb tb;

if (NULL==pszFmt

------解决思路----------------------

0==pszFmt[0]) return;

vsnprintf(logstr,MAXLINSIZE,pszFmt,argp);

ftime(&tb);

now=localtime(&tb.time);

sprintf(datestr,"%04d-%02d-%02d",now->tm_year+1900,now->tm_mon+1,now->tm_mday);

sprintf(timestr,"%02d:%02d:%02d",now->tm_hour     ,now->tm_min  ,now->tm_sec );

sprintf(mss,"%03d",tb.millitm);

printf("%s %s.%s %s",datestr,timestr,mss,logstr);

flog=fopen(logfilename1,"a");

if (NULL!=flog) {

fprintf(flog,"%s %s.%s %s",datestr,timestr,mss,logstr);

if (ftell(flog)>MAXLOGSIZE) {

fclose(flog);

if (rename(logfilename1,logfilename2)) {

remove(logfilename2);

rename(logfilename1,logfilename2);

}

} else {

fclose(flog);

}

}

}

void Log(const char *pszFmt,...) {

va_list argp;

Lock(&cs_log);

va_start(argp,pszFmt);

LogV(pszFmt,argp);

va_end(argp);

Unlock(&cs_log);

}

//Log}

#define ASIZE    200

#define BSIZE    240

#define CSIZE      2

char Abuf[ASIZE];

char Cbuf[CSIZE];

CRITICAL_SECTION cs_HEX ;

CRITICAL_SECTION cs_BBB ;

struct FIFO_BUFFER {

int  head;

int  tail;

int  size;

char data[BSIZE];

} BBB;

int No_Loop=0;

void HexDump(int cn,char *buf,int len) {

int i,j,k;

char binstr[80];

Lock(&cs_HEX);

for (i=0;i

if (0==(i%16)) {

sprintf(binstr,"%03d %04x -",cn,i);

sprintf(binstr,"%s %02x",binstr,(unsigned char)buf[i]);

} else if (15==(i%16)) {

sprintf(binstr,"%s %02x",binstr,(unsigned char)buf[i]);

sprintf(binstr,"%s  ",binstr);

for (j=i-15;j<=i;j++) {

sprintf(binstr,"%s%c",binstr,('!'

}

Log("%s\n",binstr);

} else {

sprintf(binstr,"%s %02x",binstr,(unsigned char)buf[i]);

}

}

if (0!=(i%16)) {

k=16-(i%16);

for (j=0;j

sprintf(binstr,"%s   ",binstr);

}

sprintf(binstr,"%s  ",binstr);

k=16-k;

for (j=i-k;j

sprintf(binstr,"%s%c",binstr,('!'

}

Log("%s\n",binstr);

}

Unlock(&cs_HEX);

}

int GetFromRBuf(int cn,CRITICAL_SECTION *cs,FIFO_BUFFER *fbuf,char *buf,int len) {

int lent,len1,len2;

lent=0;

Lock(cs);

if (fbuf->size>=len) {

lent=len;

if (fbuf->head+lent>BSIZE) {

len1=BSIZE-fbuf->head;

memcpy(buf     ,fbuf->data+fbuf->head,len1);

len2=lent-len1;

memcpy(buf+len1,fbuf->data           ,len2);

fbuf->head=len2;

} else {

memcpy(buf     ,fbuf->data+fbuf->head,lent);

fbuf->head+=lent;

}

fbuf->size-=lent;

}

Unlock(cs);

return lent;

}

MYVOID thdB(void *pcn) {

char        *recv_buf;

int          recv_nbytes;

int          cn;

int          wc;

int          pb;

cn=(int)pcn;

Log("%03d thdB              thread begin...\n",cn);

while (1) {

sleep_ms(10);

recv_buf=(char *)Cbuf;

recv_nbytes=CSIZE;

wc=0;

while (1) {

pb=GetFromRBuf(cn,&cs_BBB,&BBB,recv_buf,recv_nbytes);

if (pb) {

Log("%03d recv %d bytes\n",cn,pb);

HexDump(cn,recv_buf,pb);

sleep_ms(1);

} else {

sleep_ms(1000);

}

if (No_Loop) break;//

wc++;

if (wc>3600) Log("%03d %d==wc>3600!\n",cn,wc);

}

if (No_Loop) break;//

}

#ifndef WIN32

pthread_exit(NULL);

#endif

}

int PutToRBuf(int cn,CRITICAL_SECTION *cs,FIFO_BUFFER *fbuf,char *buf,int len) {

int lent,len1,len2;

Lock(cs);

lent=len;

if (fbuf->size+lent>BSIZE) {

lent=BSIZE-fbuf->size;

}

if (fbuf->tail+lent>BSIZE) {

len1=BSIZE-fbuf->tail;

memcpy(fbuf->data+fbuf->tail,buf     ,len1);

len2=lent-len1;

memcpy(fbuf->data           ,buf+len1,len2);

fbuf->tail=len2;

} else {

memcpy(fbuf->data+fbuf->tail,buf     ,lent);

fbuf->tail+=lent;

}

fbuf->size+=lent;

Unlock(cs);

return lent;

}

MYVOID thdA(void *pcn) {

char        *send_buf;

int          send_nbytes;

int          cn;

int          wc;

int           a;

int          pa;

cn=(int)pcn;

Log("%03d thdA              thread begin...\n",cn);

a=0;

while (1) {

sleep_ms(100);

memset(Abuf,a,ASIZE);

a=(a+1)%256;

if (16==a) {No_Loop=1;break;}//去掉这句可以让程序一直循环直到按Ctrl+C或Ctrl+Break或当前目录下存在文件No_Loop

send_buf=(char *)Abuf;

send_nbytes=ASIZE;

Log("%03d sending %d bytes\n",cn,send_nbytes);

HexDump(cn,send_buf,send_nbytes);

wc=0;

while (1) {

pa=PutToRBuf(cn,&cs_BBB,&BBB,send_buf,send_nbytes);

Log("%03d sent %d bytes\n",cn,pa);

HexDump(cn,send_buf,pa);

send_buf+=pa;

send_nbytes-=pa;

if (send_nbytes<=0) break;//

sleep_ms(1000);

if (No_Loop) break;//

wc++;

if (wc>3600) Log("%03d %d==wc>3600!\n",cn,wc);

}

if (No_Loop) break;//

}

#ifndef WIN32

pthread_exit(NULL);

#endif

}

int main() {

#ifdef WIN32

InitializeCriticalSection(&cs_log);

InitializeCriticalSection(&cs_HEX );

InitializeCriticalSection(&cs_BBB );

#else

pthread_t threads[2];

int threadsN;

int rc;

pthread_mutex_init(&cs_log,NULL);

pthread_mutex_init(&cs_HEX,NULL);

pthread_mutex_init(&cs_BBB,NULL);

#endif

Log("Start===========================================================\n");

BBB.head=0;

BBB.tail=0;

BBB.size=0;

#ifdef WIN32

_beginthread((void(__cdecl *)(void *))thdA,0,(void *)1);

_beginthread((void(__cdecl *)(void *))thdB,0,(void *)2);

#else

threadsN=0;

rc=pthread_create(&(threads[threadsN++]),NULL,thdA,(void *)1);if (rc) Log("%d=pthread_create %d error!\n",rc,threadsN-1);

rc=pthread_create(&(threads[threadsN++]),NULL,thdB,(void *)2);if (rc) Log("%d=pthread_create %d error!\n",rc,threadsN-1);

#endif

if (!access("No_Loop",0)) {

remove("No_Loop");

if (!access("No_Loop",0)) {

No_Loop=1;

}

}

while (1) {

sleep_ms(1000);

if (No_Loop) break;//

if (!access("No_Loop",0)) {

No_Loop=1;

}

}

sleep_ms(3000);

Log("End=============================================================\n");

#ifdef WIN32

DeleteCriticalSection(&cs_BBB );

DeleteCriticalSection(&cs_HEX );

DeleteCriticalSection(&cs_log);

#else

pthread_mutex_destroy(&cs_BBB);

pthread_mutex_destroy(&cs_HEX);

pthread_mutex_destroy(&cs_log);

#endif

return 0;

}