首页 › 线程间通信 › 网络服务端开发小结(短连接、长连接、进程池、线程池)

网络服务端开发小结(短连接、长连接、进程池、线程池)

第七根弦 2014 年 12 月 17 日 线程间通信, 进程间通信 发表评论 (0)

平时对网络编程方面比较感兴趣，看了一些相关的资料，自己也尝试写过一些不同网络模型的服务程序。这次刚好有一个新的需求，需要开发一个转发服务器。之前开发的项目，网络通讯都是处理联机交易的，网络连接都是采用短连接，这次的服务端，采用长连接的方式。

1. 轮询和主动通知选择

公司有一个客户端产品(CLIENT)，因为需要从多个客户的服务端获(SERVER)取信息，原有的设计是每个客户端通过SOCKET不断的轮询访问服务端获取信息。当CLIENT的数量不多时，轮询访问对服务端压力不大，但是当CLIENT的数量比较多时，这样的访问对服务器的压力就比较大，而且很低效。为了降低CLIENT对服务端的压力和提高信息的获取效率，采用让SERVER主动通知CLIENT的方式。不可能让每个客户SERVER都开发一个通知的服务端，所以需要一个和SERVER通讯并且把消息通知给CLIENT的转发服务端(简称TransmitServer).系统整体架构如下:

SERVER    ————–>       RECV   ——->  TransmitServer—————>CLIENT

SERVER把消息发给RECV, RECV通过消息队列发给TransmitServer，TransmitServer再把消息通知给相应的CLIENT。

2.短连接和长连接选择

因为是通知消息的方式，所以如果使用短连接的话，让TransmitServer去和ClENT主动建立连接是不可行的，那样就要让TransmitServer知道所有CLIENT的通讯地址，这是一种很笨的方式，所以不采用。当然如果使用短连接轮询的方式,让CLIENT去访问TransmitServer，TransmitServer访问SERVER，对SERVER的压力也能得到解决.

3.  进程管理方式和I/O复用的选择

网络连接处理的模型有很多，按照进程的管理方式，我分为2类，进程池和多进程(还包括线程)，以及每种方式还可以选择是否应该使用I/O复用。

这里说的进程池和多进程是，进程池是预先启动多个子进程，并且可以管理进程；多进程是指到主进程阻塞于ACCEPT处理连接请求，由子进程 单独负责每个套接口连接。

处理短连接连接，因为客户端频繁的连接服务器和断开连接，服务端的主要性能开销应该是在进程切换上，基于性能考虑采用进程池的方式会比多进程好.如果连接并发量不大,没有性能上的问题,多进程的程序会比进程池简单很多。

如果服务端处理的是长连接。如果让进程池或者多进程中一个子进程只处理一个连接,系统的主要性能开销主要取决于进程的数量,在进程的数量到一定数量的时候,会对服务器造成比较大的压力。所以在处理长连接时,一般才用I/O多路复用的方式,LINUX上I/O多路复用,有SELECT、EPOLL等。

使用I/O多路复用一方面可以让单个进程同时多个连接,可以提高并发连接数。另一方面,还可以可以让CLIENT和SERVER的通讯更加灵活,例如使用I/O多路复用,让CLIENT和SERVER可以很容易的异步通讯。

4. 线程和进程的选择

使用I/O多路复用,让单个进程可以处理多个连接,如果进程池只有一个进程, 同一个服务进程里的连接之间可以很方便的通讯,但是如果是多个子进程，那么子进程之间就不能直接通讯，通常要要消息队列、共享内存、管道等。当服务端需要处理连接之间的交互,而且性能上需要多个进程,那么使用线程池代替进程池应该是一种比较好的方式。线程和进程相比较,线程的优点是性能开销更小,因为在同一个进程空间里,线程之间的通讯很容易;因为线程共享进程空间数据,因此线程在处理的时候比进程更容易出错,线程池的方式简化了通讯方式,但是为了线程安全,这方面的复杂性就增加了。所以如果网络连接之间并不需要很多交互,每个连接处理都是独立的,那么应该选择进程。这个原则不但对网络处理,对于别的处理也是如此。

综合上面几点服务端网络模型的选择,主要是根据业务选择使用长连接或者短连接,根据连接方式。长连接一般使用线程池+I/O复用,例如网络游戏、聊天室等。短连接一般是进程池或者多进程,因为短连接不需要连接之间的交互,每个连接都是独立的,所以使用进程更加合适。

参考书目:

W.Richard Stevens <<UNIX网络编程(第一卷)>>

W.Richard Stevens <<UNIX高级环境编程>>

DEMO示例:

C++
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
#include <stdio.h>
#include <math.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <memory.h>
 
#include <sys/timeb.h>
#include <time.h>
#include <sys/time.h>
 
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include <sys/sem.h>
#include <sys/un.h>
#include <sys/socket.h>
 
#include <netinet/in.h>
#include <signal.h>
#include <netdb.h>
 
#define SETSIZE 3000
#define MAXLINE 1000
#define LISTENQ 2000
#define CHILDMAX 0
#define NMERIDLEN 6
#define NUSRIDLEN 40
#define NHANDHEADLEN 12
#define NTESTPACKETLEN 1
#define NPACKETHEADLEN 4
#define NMAXPACKETLEN 256
#define TICKTIME 200000
#define TCP_CLOSE -101
#define TCP_PACKET_AVILD -102
int children[ CHILDMAX];
int ChildNum;
 
typedef struct STRU_LINK_LIST
{
char saMerId[ NMERIDLEN + 1];
char saUsrId[ NUSRIDLEN + 1];
int socket;
} STRULINKLIST;
 
STRULINKLIST client[ SETSIZE];
 
typedef struct STRU_HAND_PACKET
{
char saMerId[ NMERIDLEN + 1];
char saUsrId[ NUSRIDLEN + 1];
char saHead[ NHANDHEADLEN + 1];
} STRUHANDPACKET;
 
typedef struct STRU_TEST_PACKET
{
char saPacket[ NTESTPACKETLEN + 1];
} STRUTESTPACKET;
 
typedef struct STRU_NOTIFY_PACKET
{
char saMerId[ NMERIDLEN + 1];
char saUsrId[ NUSRIDLEN + 1];
} STRUNOTIFYPACKET;
 
void sig_do(int signo);
void DeleteLink(STRULINKLIST* link, fd_set* allset);
void ProcessCmd(char* line, int len, STRULINKLIST* link, fd_set* allset,
int maxi);
int pack_packet(char cmd, void* stru, char* packet);
void get_packet(char cmd, char* packet, void *stru);
 
int main(int argc, char **argv)
{
int i, maxi, maxfd, listenfd, connfd, sockfd;
fd_set rset, allset, conset;
socklen_t clilen;
struct sockaddr_in cliaddr, servaddr;
int nready;
ssize_t n;
char line[MAXLINE];
int pid;
struct timeval timeval;
 
listenfd = socket( AF_INET, SOCK_STREAM, IPPROTO_TCP);
 
bzero(&servaddr, sizeof(servaddr));
servaddr.sin_family = AF_INET;
servaddr.sin_addr.s_addr = htonl(INADDR_ANY);
servaddr.sin_port = htons(9999);
 
if (bind(listenfd, (struct sockaddr *) &servaddr, sizeof(servaddr)) < 0)
{
printf("bind error/n");
exit(-1);
}
listen(listenfd, 2000);
signal( SIGINT, sig_do);
ProcessSelect(listenfd);
}
 
Tick( )
{
usleep( TICKTIME );
}
 
void ProcessSelect(int listen_fd)
{
int i, maxi, maxfd, listenfd, connfd, sockfd;
fd_set rset, allset, conset;
socklen_t clilen;
char line[MAXLINE];
char saTmpBuf[128];
int nready;
ssize_t n, nLen;
struct sockaddr_in cliaddr;
struct timeval timeval;
 
timeval.tv_sec = 0;
timeval.tv_usec = 0;
 
listenfd = listen_fd;
maxfd = -1;
maxi = -1;
for (i = 0; i < SETSIZE; i++)
client[i].socket = -1;
FD_ZERO(&allset);
FD_ZERO(&conset);
FD_SET(listenfd, &conset);
 
for (;;)
{
Tick();
rset = conset; /* structure assignment */
nready = select(listenfd + 1, &rset, NULL, NULL, &timeval);
/*accept新的连接*/
 
nready = select(listenfd + 1, &rset, NULL, NULL, &timeval);
/*accept新的连接*/
if (FD_ISSET(listenfd, &rset))
{ /* new client connection */
clilen = sizeof(cliaddr);
connfd = accept(listenfd, (struct sockaddr *) &cliaddr, &clilen);
 
if (connfd > 0)
{
ErrorLog(ERROR,
"pid:%d accept new client: %s, port %d socket[%d]",
getpid(), inet_ntoa(&cliaddr.sin_addr),
ntohs(cliaddr.sin_port), connfd);
 
for (i = 0; i < SETSIZE; i++)
if (client[i].socket < 0)
{
/* 保存新的连接*/
client[i].socket = connfd;
break;
}
if (i == SETSIZE)
{
close(connfd);
ErrorLog(ERROR, "link number max");
}
else
{
/* 添加新的连接到集合 */
FD_SET(connfd, &allset);
if (connfd > maxfd)
maxfd = connfd; /* for select */
 
if (connfd > maxfd)
maxfd = connfd; /* for select */
/*记录client的最大下标*/
if (i > maxi)
maxi = i;
}
}
else
ErrorLog(ERROR, "accept error ");
/*
if (--nready <= 0)
continue;
*/
}
rset = allset;
nready = select(maxfd + 1, &rset, NULL, NULL, &timeval);
for (i = 0; i <= maxi; i++)
{
memset(line, 0, sizeof(line));
if ((sockfd = client[i].socket) < 0)
continue;
if (FD_ISSET(sockfd, &rset))
{
if ((nLen = RecvPacket(sockfd, line)) <= 0)
{
DeleteLink(&client[i], &allset);
}
else
{
/*处理客户端的请求命令*/
ProcessCmd(line, nLen, &client[i], &allset, maxi);
 
/*处理客户端的请求命令*/
ProcessCmd(line, nLen, &client[i], &allset, maxi);
}
if (--nready <= 0)
break;
}
}
/*读取消息队列数据*/
 
/*发送网站数据*/
/*
for( i = 0; i < 2; i++ )
{
memset( line, 0, sizeof( line ));
sprintf( line, "N%06dWUDY",i);
nLen = strlen(line);
ProcessCmd( line, nLen, NULL , &allset, maxi);
}
*/
/*发送测试包*/
/*
memset( line, 0, sizeof( line ));
strcpy( line, "T");
ProcessCmd( line, 1, NULL , &allset, maxi);
*/
/*转发数据*/
 
}
 
}
 
int make_child()
{
int pid;
pid = fork();
if (pid < 0)
{
printf("fork error/n");
}
else if (pid == 0)
{
/* do child*/
}
return pid;
}
 
int RecvPacket(int sockfd, char* packet)
{
int n, nPacketLen;
char saTmpBuf[ NPACKETHEADLEN + 1];
 
if ((n = read(sockfd, saTmpBuf, NPACKETHEADLEN)) == 0)
{
return TCP_CLOSE;
}
else
{
// process cmd for client
 
nPacketLen = atoi(saTmpBuf);
if ((n = read(sockfd, packet, nPacketLen)) <= 0)
{
return TCP_CLOSE;
}
if (n < nPacketLen)
{
ErrorLog(ERROR, "报文长度不符");
return TCP_PACKET_AVILD;
}
}
return n;
}
 
void DeleteLink(STRULINKLIST* link, fd_set* allset)
{
close(link->socket);
FD_CLR(link->socket, allset);
link->socket = -1;
}
 
void HandLink(STRULINKLIST* link, STRUHANDPACKET handPacket)
{
memset(link->saMerId, 0, sizeof(link->saMerId));
memset(link->saUsrId, 0, sizeof(link->saUsrId));
memcpy(link->saMerId, handPacket.saMerId, NMERIDLEN);
memcpy(link->saUsrId, handPacket.saUsrId, NMERIDLEN);
}
 
void ProcessCmd(char* line, int len, STRULINKLIST* link, fd_set* pset, int maxi)
{
char saTmpCmd[2];
char saPacket[ MAXLINE];
int i;
STRUHANDPACKET StruHandPacket;
STRUTESTPACKET StruTestPacket;
STRUNOTIFYPACKET StruNotifyPacket;
 
memset(saTmpCmd, 0, sizeof(saTmpCmd));
memset(saPacket, 0, sizeof(saPacket));
memcpy(saTmpCmd, line, 1);
memcpy(saPacket, line + 1, len - 1);
 
switch (saTmpCmd[0])
{
case 'H':
get_packet(saTmpCmd[0], saPacket, &StruHandPacket);
HandLink(link, StruHandPacket);
break;
case 'C':
break;
case 'T':
 
get_packet(saTmpCmd[0], saPacket, &StruTestPacket);
memset(saPacket, 0, sizeof(saPacket));
len = pack_packet(saTmpCmd[0], &StruTestPacket, saPacket);
for (i = 0; i <= maxi; i++)
{
 
{
int sockfd;
int nLen;
sockfd = client[i].socket;
 
if (write(sockfd, saPacket, len) < 0)
{
/*发送失败，关闭连接*/
DeleteLink(&client[i], pset);
continue;
}
}
break;
case 'N':
get_packet(saTmpCmd[0], saPacket, &StruNotifyPacket);
memset(saPacket, 0, sizeof(saPacket));
len = pack_packet(saTmpCmd[0], &StruNotifyPacket, saPacket);
for (i = 0; i <= maxi; i++)
{
if (!strcmp(client[i].saMerId, StruNotifyPacket.saMerId)
&& !strcmp(client[i].saUsrId, StruNotifyPacket.saUsrId)
&& client[i].socket != -1)
{
if (write(client[i].socket, saPacket, len) < 0)
{
/*发送失败，关闭连接*/
 
/*发送失败，关闭连接*/
DeleteLink(&client[i], pset);
continue;
}
}
}
break;
default:
break;
}
}
 
int pack_packet( char cmd, void* stru, char* packet )
{
STRUNOTIFYPACKET StruNotifyPacket;
STRUTESTPACKET StruTestPacket;
int nLen;
 
switch( cmd )
{
case 'H':
break;
case 'C':
break;
case 'T':
memset( &StruTestPacket, 0, sizeof( STRUTESTPACKET ));
memcpy( &StruTestPacket,(STRUTESTPACKET*)stru , sizeof( STRUTESTPACKET ) );
nLen = 1 + strlen(StruTestPacket.saPacket );
sprintf( packet, "%04d%c%s%s", nLen,cmd, StruTestPacket.saPacket);
break;
case 'N':
memset( &StruNotifyPacket, 0, sizeof( STRUNOTIFYPACKET ));
 
memcpy( &StruNotifyPacket, (STRUNOTIFYPACKET*)stru, sizeof(STRUNOTIFYPACKET));
 
nLen = strlen("N") + strlen( StruNotifyPacket.saMerId ) + strlen( StruNotifyPacket.saUsrId );
sprintf( packet, "%04d%c%s%s", nLen,cmd, StruNotifyPacket.saMerId, StruNotifyPacket.saUsrId );
break;
default:
 
break;
}
return nLen+NPACKETHEADLEN;
}
 
void get_packet( char cmd, char* packet, void *stru )
{
char saPacket[ NMAXPACKETLEN+1 ];
int nLen;
nLen = strlen( packet );
switch( cmd )
{
 
case 'H':
memset( (STRUHANDPACKET*)stru, 0, sizeof( STRUHANDPACKET ));
memcpy( ((STRUHANDPACKET*)stru)->saHead, packet, NHANDHEADLEN );
memcpy( ((STRUHANDPACKET*)stru)->saMerId, packet+NHANDHEADLEN, NMERIDLEN );
memcpy( ((STRUHANDPACKET*)stru)->saUsrId, packet+NHANDHEADLEN+NMERIDLEN, nLen-NHANDHEADLEN-NMERIDLEN);
break;
case 'C':
break;
case 'T':
memset( (STRUTESTPACKET*)stru, 0, sizeof( STRUTESTPACKET ));
memcpy( ((STRUTESTPACKET*)stru)->saPacket, packet, NTESTPACKETLEN );
break;
case 'N':
memset( (STRUNOTIFYPACKET*)stru, 0, sizeof( STRUNOTIFYPACKET ));
 
memcpy( ((STRUNOTIFYPACKET*)stru)->saMerId, packet, NMERIDLEN );
memcpy( ((STRUNOTIFYPACKET*)stru)->saUsrId, packet+NMERIDLEN,nLen-NMERIDLEN );
break;
default:
break;
 
}
}
 
void sig_do( int signo )
{
int i;
for( i = 0; i < ChildNum; i++ )
kill(children[i] , SIGTERM);
while( wait(NULL) > 0 )
;
if( errno != ECHILD )
printf( "wait error/n" );
exit(0);
}

, , , |

发表评论