21 个解决方案
#1
udp 数据是不可靠的 所以接收速度要快. 接收的数据可以先不用处理.
启动一个子线程 只管接收数据 ,然后 放入 一个队列中 (或者链表 都可以) 然后启动多个子线程 来处理.
这样就不会丢失数据了.
启动一个子线程 只管接收数据 ,然后 放入 一个队列中 (或者链表 都可以) 然后启动多个子线程 来处理.
这样就不会丢失数据了.
#2
接收，放入缓存（内存队列等），然后用几个线程处理
#3
数据处理重开一个进程
#4
您好，我一开始思路是这样的，一个线程接收数据放入一个循环缓冲区，当数据到达一定量后再触发运算线程取走数据并进行运算，可是在运算这段时间由于设置了临界保护区所以接受部分线程是不能写数据的。我的疑问是如何让接受线程不停处理线程也能处理接受的数据，两个线程同时操作一个队列不会出问题吗？
#5
但是处理速度肯还是更不上接受速度的，这样怎么处理呢？
#6
几个线程如何同时操作一个队列里的数据呢？这个具体我不太清楚望指教，谢谢！
#7
首先 你要明白队列的含义. 就是数据等待处理 在排队
第二 要处理队列的数据 只需要把数据从队列中取出来 就可以去处理了, 多线程取出的过程 需要加锁.
或者 你可以使用epool 模型 来接收和处理
#8
我用了一个 内存队列 和 redis 队列 处理过秒100万的udp 数据接收. 测试一个包都不丢(程序丢包)
使用了分布式处理方式.
一台服务器 专门接收udp 数据包,然后先保存在内存list 中 每10毫秒 批量插入redis 队列
另一台(或N台)服务器开启多个线程 从redis 队列中 取出数据 去处理.这样 只要接收服务器 能承受, 随时可以 配置多台处理的服务器,每个服务器的线程数也是可以通过配置来控制.
使用了分布式处理方式.
一台服务器 专门接收udp 数据包,然后先保存在内存list 中 每10毫秒 批量插入redis 队列
另一台(或N台)服务器开启多个线程 从redis 队列中 取出数据 去处理.这样 只要接收服务器 能承受, 随时可以 配置多台处理的服务器,每个服务器的线程数也是可以通过配置来控制.
#9
这是我的写的部分代码，但这样速度完全达不到，请问是什么原因呢？
//数据接收线程
UINT ReciveThread(LPVOID params)
{
OutputDebugString(">>>> Start ReceiveThread...\r\n");
CDTUDlg *pDlg = (CDTUDlg *)params;
int err = 0;
int sockaddrlen = 0;
sockaddrlen = sizeof(pDlg->s_RemoteAddr);
while(theApp.m_bLooping)
{
//通过udp接收数据
OutputDebugString(">>>> ReciveThread WaitForSingleObject : m_hEvent_ProcessingEnd\r\n");
//::WaitForSingleObject(theApp.m_hEvent_ProcessingEnd,INFINITE);
OutputDebugString("recvfrom...\r\n");
for(int i=0;i<10;i++)
{
err = recvfrom( //返回正确接收的字节数
pDlg->UDPSocket, /* receive UDP data */
pDlg->RecMsg.msg,
sizeof(pDlg->RecMsg.msg),
0,
(sockaddr *)&pDlg->s_RemoteAddr,
(int *)&sockaddrlen
);
if (err == SOCKET_ERROR) /* Is receive data right */
{
return 0;
}
pDlg->RecMsg.len = err;
//互斥的访问缓冲区
for(int i=0;i<pDlg->RecMsg.len;i++)//将数据放入缓存
{
EnterCriticalSection(&theApp.g_cs);
theApp.dataBuffer.push(pDlg->RecMsg.msg[i]);
LeaveCriticalSection(&theApp.g_cs);
}
}
Sleep(1);
}
return 1;
}
//数据处理线程
UINT ProcessDataThread(LPVOID params)
{
OutputDebugString(">>>> Start ProcessDataThread...\r\n");
CDTUDlg *pDlg = (CDTUDlg *)params;//point to Dlg
while(theApp.m_bLooping)
{
for(int i=0;i<10;i++)
{
if(theApp.dataBuffer.size()>=4114)
{
EnterCriticalSection(&theApp.g_cs);
pDlg->get();//解析数据
LeaveCriticalSection(&theApp.g_cs);
pDlg->RecHandle();//处理数据
}
}
return 0;
}
#10
如果的确是这样，如果不是程序设计的问题——就是CPU真的不够用了，增加硬件吧，
#11
可以帮忙看下代码么?
#12
做压力测试的话 把你的Sleep(1)和OutputDebugString 都去掉 最耗时的地方 就是IO操作
而Sleep(1) 在windows 下是1毫秒 1秒钟最多接收1000条数据
linux 上是微秒级的 也会因为这个操作影响接收速度.
而Sleep(1) 在windows 下是1毫秒 1秒钟最多接收1000条数据
linux 上是微秒级的 也会因为这个操作影响接收速度.
#13
哦哦！，请问c++下可不可以不加锁的情况下实现同时对队列读写操作呢？因为我的读取数据处理那块很费时如果加锁会影响接受速度而造成丢包
#14
不需要这么复杂，双缓冲区接收数据，处理线程设置一个标志表明正在处理哪个缓冲区，接收线程读这个标志，向空闲的那个缓冲区写数据就可以了。
#15
这个是假的吧 ？？？
#16
加锁解锁本身的动作，是很快的（尤其是对于你说的每秒4M的数据来不及处理）。
你说的读取数据处理很费时而影响到接收速度，明显还是逻辑处理的问题，还是单线程的思路。
看起来你接收到一块缓存，然后用另外的线程加锁处理这一块缓存，然而接收线程由于这块缓存被锁而无法接收。这和单线程有啥区别。
这肯定不行。要用缓存队列或者链表之类的。
接收一块缓存，加锁，入队列，解锁。然后接收新new的缓存。
多个处理线程加锁，从队列中取一块缓存，解锁，处理数据。delete缓存，再重复。
当然，涉及到多个处理线程如何正确的分包，粘包，这是另外的事情了。
你说的读取数据处理很费时而影响到接收速度，明显还是逻辑处理的问题，还是单线程的思路。
看起来你接收到一块缓存，然后用另外的线程加锁处理这一块缓存，然而接收线程由于这块缓存被锁而无法接收。这和单线程有啥区别。
这肯定不行。要用缓存队列或者链表之类的。
接收一块缓存，加锁，入队列，解锁。然后接收新new的缓存。
多个处理线程加锁，从队列中取一块缓存，解锁，处理数据。delete缓存，再重复。
当然，涉及到多个处理线程如何正确的分包，粘包，这是另外的事情了。
#17
仅供参考：
//循环向a函数每次发送200个字节长度（这个是固定的）的buffer,
//a函数中需要将循环传进来的buffer，组成240字节（也是固定的）的新buffer进行处理，
//在处理的时候每次从新buffer中取两个字节打印
#ifdef _MSC_VER
#pragma warning(disable:4996)
#endif
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#ifdef _MSC_VER
#include <windows.h>
#include <process.h>
#include <io.h>
#define MYVOID void
#define vsnprintf _vsnprintf
#else
#include <unistd.h>
#include <sys/time.h>
#include <pthread.h>
#define CRITICAL_SECTION pthread_mutex_t
#define MYVOID void *
#endif
//Log{
#define MAXLOGSIZE 20000000
#define MAXLINSIZE 16000
#include <time.h>
#include <sys/timeb.h>
#include <stdarg.h>
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 _MSC_VER
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<len;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,('!'<buf[j]&&buf[j]<='~')?buf[j]:'.');
}
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<k;j++) {
sprintf(binstr,"%s ",binstr);
}
sprintf(binstr,"%s ",binstr);
k=16-k;
for (j=i-k;j<i;j++) {
sprintf(binstr,"%s%c",binstr,('!'<buf[j]&&buf[j]<='~')?buf[j]:'.');
}
Log("%s\n",binstr);
}
Unlock(&cs_HEX);
}
int GetFromRBuf(int cn,CRITICAL_SECTION *cs,struct 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 _MSC_VER
pthread_exit(NULL);
#endif
}
int PutToRBuf(int cn,CRITICAL_SECTION *cs,struct 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 _MSC_VER
pthread_exit(NULL);
#endif
}
int main() {
#ifdef _MSC_VER
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 _MSC_VER
_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 _MSC_VER
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;
}
#18
空间与时间的问题.
你要处理快, 那么就只能多占缓存空间.
要少缓存空间, 那么必然处理就慢了.
可以开启一个线程, 专门收数据, 收到数据放到缓冲区中或者队列中, 这个过程应该是非常快的, 肯定会快过一般的网络传输速度.
然后再开一个线程来专门处理数据, 数据处理一般会比较慢.
这样一来, 你就要权衡了, 因为生产得快, 消费得慢, 必然导致内存占用疯涨. 所以变成时间与空间的问题了.
你要处理快, 那么就只能多占缓存空间.
要少缓存空间, 那么必然处理就慢了.
可以开启一个线程, 专门收数据, 收到数据放到缓冲区中或者队列中, 这个过程应该是非常快的, 肯定会快过一般的网络传输速度.
然后再开一个线程来专门处理数据, 数据处理一般会比较慢.
这样一来, 你就要权衡了, 因为生产得快, 消费得慢, 必然导致内存占用疯涨. 所以变成时间与空间的问题了.
#19
这是当时在生产用的策略 ,真假自知!
#20
任何收发两端速度不一致的通讯，都需要在它们之间使用一个足够大的FIFO缓冲区。
对任何FIFO缓冲区的使用，都需要仔细考虑接收端接收时超时无数据和发送端发送时FIFO缓冲区已满这两种情况下该如何做。
这些概念都在这段经典代码中有所体现。
这段经典代码还包括以下必须考虑的因素：
◆跨Windows和Linux平台
◆多线程锁
◆多线程日志
◆日志文件占用的磁盘空间可控
◆日志中的时间包括毫秒
◆传输的数据对应的每个字节到底是几
◆如何退出多线程程序
◆……
#21
这个方案到时真的，这个100万的数据是指100万条UDP消息么 ？？
#1
udp 数据是不可靠的 所以接收速度要快. 接收的数据可以先不用处理.
启动一个子线程 只管接收数据 ,然后 放入 一个队列中 (或者链表 都可以) 然后启动多个子线程 来处理.
这样就不会丢失数据了.
启动一个子线程 只管接收数据 ,然后 放入 一个队列中 (或者链表 都可以) 然后启动多个子线程 来处理.
这样就不会丢失数据了.
#2
接收，放入缓存（内存队列等），然后用几个线程处理
#3
数据处理重开一个进程
#4
您好，我一开始思路是这样的，一个线程接收数据放入一个循环缓冲区，当数据到达一定量后再触发运算线程取走数据并进行运算，可是在运算这段时间由于设置了临界保护区所以接受部分线程是不能写数据的。我的疑问是如何让接受线程不停处理线程也能处理接受的数据，两个线程同时操作一个队列不会出问题吗？
#5
但是处理速度肯还是更不上接受速度的，这样怎么处理呢？
#6
几个线程如何同时操作一个队列里的数据呢？这个具体我不太清楚望指教，谢谢！
#7
首先 你要明白队列的含义. 就是数据等待处理 在排队
第二 要处理队列的数据 只需要把数据从队列中取出来 就可以去处理了, 多线程取出的过程 需要加锁.
或者 你可以使用epool 模型 来接收和处理
#8
我用了一个 内存队列 和 redis 队列 处理过秒100万的udp 数据接收. 测试一个包都不丢(程序丢包)
使用了分布式处理方式.
一台服务器 专门接收udp 数据包,然后先保存在内存list 中 每10毫秒 批量插入redis 队列
另一台(或N台)服务器开启多个线程 从redis 队列中 取出数据 去处理.这样 只要接收服务器 能承受, 随时可以 配置多台处理的服务器,每个服务器的线程数也是可以通过配置来控制.
使用了分布式处理方式.
一台服务器 专门接收udp 数据包,然后先保存在内存list 中 每10毫秒 批量插入redis 队列
另一台(或N台)服务器开启多个线程 从redis 队列中 取出数据 去处理.这样 只要接收服务器 能承受, 随时可以 配置多台处理的服务器,每个服务器的线程数也是可以通过配置来控制.
#9
这是我的写的部分代码，但这样速度完全达不到，请问是什么原因呢？
//数据接收线程
UINT ReciveThread(LPVOID params)
{
OutputDebugString(">>>> Start ReceiveThread...\r\n");
CDTUDlg *pDlg = (CDTUDlg *)params;
int err = 0;
int sockaddrlen = 0;
sockaddrlen = sizeof(pDlg->s_RemoteAddr);
while(theApp.m_bLooping)
{
//通过udp接收数据
OutputDebugString(">>>> ReciveThread WaitForSingleObject : m_hEvent_ProcessingEnd\r\n");
//::WaitForSingleObject(theApp.m_hEvent_ProcessingEnd,INFINITE);
OutputDebugString("recvfrom...\r\n");
for(int i=0;i<10;i++)
{
err = recvfrom( //返回正确接收的字节数
pDlg->UDPSocket, /* receive UDP data */
pDlg->RecMsg.msg,
sizeof(pDlg->RecMsg.msg),
0,
(sockaddr *)&pDlg->s_RemoteAddr,
(int *)&sockaddrlen
);
if (err == SOCKET_ERROR) /* Is receive data right */
{
return 0;
}
pDlg->RecMsg.len = err;
//互斥的访问缓冲区
for(int i=0;i<pDlg->RecMsg.len;i++)//将数据放入缓存
{
EnterCriticalSection(&theApp.g_cs);
theApp.dataBuffer.push(pDlg->RecMsg.msg[i]);
LeaveCriticalSection(&theApp.g_cs);
}
}
Sleep(1);
}
return 1;
}
//数据处理线程
UINT ProcessDataThread(LPVOID params)
{
OutputDebugString(">>>> Start ProcessDataThread...\r\n");
CDTUDlg *pDlg = (CDTUDlg *)params;//point to Dlg
while(theApp.m_bLooping)
{
for(int i=0;i<10;i++)
{
if(theApp.dataBuffer.size()>=4114)
{
EnterCriticalSection(&theApp.g_cs);
pDlg->get();//解析数据
LeaveCriticalSection(&theApp.g_cs);
pDlg->RecHandle();//处理数据
}
}
return 0;
}
#10
如果的确是这样，如果不是程序设计的问题——就是CPU真的不够用了，增加硬件吧，
#11
可以帮忙看下代码么?
#12
做压力测试的话 把你的Sleep(1)和OutputDebugString 都去掉 最耗时的地方 就是IO操作
而Sleep(1) 在windows 下是1毫秒 1秒钟最多接收1000条数据
linux 上是微秒级的 也会因为这个操作影响接收速度.
而Sleep(1) 在windows 下是1毫秒 1秒钟最多接收1000条数据
linux 上是微秒级的 也会因为这个操作影响接收速度.
#13
哦哦！，请问c++下可不可以不加锁的情况下实现同时对队列读写操作呢？因为我的读取数据处理那块很费时如果加锁会影响接受速度而造成丢包
#14
不需要这么复杂，双缓冲区接收数据，处理线程设置一个标志表明正在处理哪个缓冲区，接收线程读这个标志，向空闲的那个缓冲区写数据就可以了。
#15
这个是假的吧 ？？？
#16
加锁解锁本身的动作，是很快的（尤其是对于你说的每秒4M的数据来不及处理）。
你说的读取数据处理很费时而影响到接收速度，明显还是逻辑处理的问题，还是单线程的思路。
看起来你接收到一块缓存，然后用另外的线程加锁处理这一块缓存，然而接收线程由于这块缓存被锁而无法接收。这和单线程有啥区别。
这肯定不行。要用缓存队列或者链表之类的。
接收一块缓存，加锁，入队列，解锁。然后接收新new的缓存。
多个处理线程加锁，从队列中取一块缓存，解锁，处理数据。delete缓存，再重复。
当然，涉及到多个处理线程如何正确的分包，粘包，这是另外的事情了。
你说的读取数据处理很费时而影响到接收速度，明显还是逻辑处理的问题，还是单线程的思路。
看起来你接收到一块缓存，然后用另外的线程加锁处理这一块缓存，然而接收线程由于这块缓存被锁而无法接收。这和单线程有啥区别。
这肯定不行。要用缓存队列或者链表之类的。
接收一块缓存，加锁，入队列，解锁。然后接收新new的缓存。
多个处理线程加锁，从队列中取一块缓存，解锁，处理数据。delete缓存，再重复。
当然，涉及到多个处理线程如何正确的分包，粘包，这是另外的事情了。
#17
仅供参考：
//循环向a函数每次发送200个字节长度（这个是固定的）的buffer,
//a函数中需要将循环传进来的buffer，组成240字节（也是固定的）的新buffer进行处理，
//在处理的时候每次从新buffer中取两个字节打印
#ifdef _MSC_VER
#pragma warning(disable:4996)
#endif
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#ifdef _MSC_VER
#include <windows.h>
#include <process.h>
#include <io.h>
#define MYVOID void
#define vsnprintf _vsnprintf
#else
#include <unistd.h>
#include <sys/time.h>
#include <pthread.h>
#define CRITICAL_SECTION pthread_mutex_t
#define MYVOID void *
#endif
//Log{
#define MAXLOGSIZE 20000000
#define MAXLINSIZE 16000
#include <time.h>
#include <sys/timeb.h>
#include <stdarg.h>
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 _MSC_VER
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<len;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,('!'<buf[j]&&buf[j]<='~')?buf[j]:'.');
}
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<k;j++) {
sprintf(binstr,"%s ",binstr);
}
sprintf(binstr,"%s ",binstr);
k=16-k;
for (j=i-k;j<i;j++) {
sprintf(binstr,"%s%c",binstr,('!'<buf[j]&&buf[j]<='~')?buf[j]:'.');
}
Log("%s\n",binstr);
}
Unlock(&cs_HEX);
}
int GetFromRBuf(int cn,CRITICAL_SECTION *cs,struct 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 _MSC_VER
pthread_exit(NULL);
#endif
}
int PutToRBuf(int cn,CRITICAL_SECTION *cs,struct 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 _MSC_VER
pthread_exit(NULL);
#endif
}
int main() {
#ifdef _MSC_VER
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 _MSC_VER
_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 _MSC_VER
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;
}
#18
空间与时间的问题.
你要处理快, 那么就只能多占缓存空间.
要少缓存空间, 那么必然处理就慢了.
可以开启一个线程, 专门收数据, 收到数据放到缓冲区中或者队列中, 这个过程应该是非常快的, 肯定会快过一般的网络传输速度.
然后再开一个线程来专门处理数据, 数据处理一般会比较慢.
这样一来, 你就要权衡了, 因为生产得快, 消费得慢, 必然导致内存占用疯涨. 所以变成时间与空间的问题了.
你要处理快, 那么就只能多占缓存空间.
要少缓存空间, 那么必然处理就慢了.
可以开启一个线程, 专门收数据, 收到数据放到缓冲区中或者队列中, 这个过程应该是非常快的, 肯定会快过一般的网络传输速度.
然后再开一个线程来专门处理数据, 数据处理一般会比较慢.
这样一来, 你就要权衡了, 因为生产得快, 消费得慢, 必然导致内存占用疯涨. 所以变成时间与空间的问题了.
#19
这是当时在生产用的策略 ,真假自知!
#20
任何收发两端速度不一致的通讯，都需要在它们之间使用一个足够大的FIFO缓冲区。
对任何FIFO缓冲区的使用，都需要仔细考虑接收端接收时超时无数据和发送端发送时FIFO缓冲区已满这两种情况下该如何做。
这些概念都在这段经典代码中有所体现。
这段经典代码还包括以下必须考虑的因素：
◆跨Windows和Linux平台
◆多线程锁
◆多线程日志
◆日志文件占用的磁盘空间可控
◆日志中的时间包括毫秒
◆传输的数据对应的每个字节到底是几
◆如何退出多线程程序
◆……
#21
这个方案到时真的，这个100万的数据是指100万条UDP消息么 ？？