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
| #define _GNU_SOURCE
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <poll.h>
#include <time.h>
/**
* 高性能服务器结构
*/
typedef struct {
int server_fd;
int port;
struct pollfd *clients;
int max_clients;
int client_count;
unsigned long messages_sent;
unsigned long bytes_sent;
} high_perf_server_t;
/**
* 初始化高性能服务器
*/
int server_init(high_perf_server_t *server, int port, int max_clients) {
struct sockaddr_in server_addr;
memset(server, 0, sizeof(high_perf_server_t));
server->port = port;
server->max_clients = max_clients;
server->client_count = 0;
server->messages_sent = 0;
server->bytes_sent = 0;
// 分配客户端数组
server->clients = calloc(max_clients + 1, sizeof(struct pollfd));
if (!server->clients) {
perror("分配客户端数组失败");
return -1;
}
// 创建服务器套接字
server->server_fd = socket(AF_INET, SOCK_STREAM, 0);
if (server->server_fd == -1) {
perror("创建服务器套接字失败");
free(server->clients);
return -1;
}
// 设置套接字选项
int opt = 1;
if (setsockopt(server->server_fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) == -1) {
perror("设置套接字选项失败");
close(server->server_fd);
free(server->clients);
return -1;
}
// 设置服务器地址
memset(&server_addr, 0, sizeof(server_addr));
server_addr.sin_family = AF_INET;
server_addr.sin_addr.s_addr = INADDR_ANY;
server_addr.sin_port = htons(port);
// 绑定套接字
if (bind(server->server_fd, (struct sockaddr*)&server_addr, sizeof(server_addr)) == -1) {
perror("绑定套接字失败");
close(server->server_fd);
free(server->clients);
return -1;
}
// 监听连接
if (listen(server->server_fd, 10) == -1) {
perror("监听失败");
close(server->server_fd);
free(server->clients);
return -1;
}
// 设置服务器套接字为poll监听
server->clients[0].fd = server->server_fd;
server->clients[0].events = POLLIN;
printf("高性能服务器初始化完成，监听端口 %d\n", port);
return 0;
}
/**
* 接受新客户端连接
*/
int server_accept_client(high_perf_server_t *server) {
struct sockaddr_in client_addr;
socklen_t client_len = sizeof(client_addr);
int client_fd;
client_fd = accept(server->server_fd, (struct sockaddr*)&client_addr, &client_len);
if (client_fd == -1) {
perror("接受连接失败");
return -1;
}
if (server->client_count >= server->max_clients) {
printf("客户端数量已达上限，拒绝连接\n");
close(client_fd);
return -1;
}
// 添加到客户端数组
int index = server->client_count + 1;
server->clients[index].fd = client_fd;
server->clients[index].events = POLLOUT; // 准备发送数据
server->client_count++;
printf("新客户端连接: %s:%d (fd=%d)\n",
inet_ntoa(client_addr.sin_addr), ntohs(client_addr.sin_port), client_fd);
return 0;
}
/**
* 使用sendmmsg向客户端批量发送消息
*/
int server_send_batch_messages(high_perf_server_t *server, int client_index) {
int client_fd = server->clients[client_index].fd;
const int BATCH_SIZE = 8; // 每次批量发送8个消息
struct mmsghdr msgvec[BATCH_SIZE];
struct iovec iov[BATCH_SIZE][1];
char messages[BATCH_SIZE][128];
int messages_to_send = 0;
// 准备批量发送消息
memset(msgvec, 0, sizeof(msgvec));
for (int i = 0; i < BATCH_SIZE; i++) {
// 构造消息内容
snprintf(messages[i], sizeof(messages[i]),
"Server Message %lu: Batch %d, Index %d",
server->messages_sent + i + 1,
(int)(server->messages_sent / BATCH_SIZE) + 1, i + 1);
// 设置缓冲区
iov[i][0].iov_base = messages[i];
iov[i][0].iov_len = strlen(messages[i]);
// 设置消息头
msgvec[i].msg_hdr.msg_iov = iov[i];
msgvec[i].msg_hdr.msg_iovlen = 1;
msgvec[i].msg_hdr.msg_name = NULL;
msgvec[i].msg_hdr.msg_namelen = 0;
messages_to_send++;
}
// 批量发送消息
int messages_sent = sendmmsg(client_fd, msgvec, messages_to_send, MSG_NOSIGNAL);
if (messages_sent == -1) {
if (errno == EPIPE || errno == ECONNRESET) {
printf("客户端 %d 连接断开\n", client_fd);
return -1;
}
perror("sendmmsg 失败");
return -1;
}
// 更新统计信息
server->messages_sent += messages_sent;
for (int i = 0; i < messages_sent; i++) {
server->bytes_sent += msgvec[i].msg_len;
}
printf("向客户端 %d 批量发送 %d 个消息\n", client_fd, messages_sent);
return 0;
}
/**
* 运行服务器主循环
*/
int server_run(high_perf_server_t *server) {
printf("服务器开始运行，等待客户端连接...\n");
time_t start_time = time(NULL);
while (difftime(time(NULL), start_time) < 30) { // 运行30秒
// 使用poll等待事件
int nfds = server->client_count + 1;
int activity = poll(server->clients, nfds, 1000); // 1秒超时
if (activity == -1) {
if (errno == EINTR) continue; // 被信号中断
perror("poll 失败");
break;
}
if (activity == 0) {
// 超时，继续循环
continue;
}
// 检查服务器套接字（新连接）
if (server->clients[0].revents & POLLIN) {
server_accept_client(server);
activity--;
}
// 检查客户端套接字（准备发送数据）
for (int i = 1; i <= server->client_count && activity > 0; i++) {
if (server->clients[i].revents & POLLOUT) {
if (server_send_batch_messages(server, i) == -1) {
// 客户端断开连接，移除客户端
close(server->clients[i].fd);
// 将最后一个客户端移到当前位置
if (i < server->client_count) {
server->clients[i] = server->clients[server->client_count];
}
server->client_count--;
i--; // 重新检查当前位置
}
activity--;
}
}
}
return 0;
}
/**
* 清理服务器资源
*/
void server_cleanup(high_perf_server_t *server) {
// 关闭所有客户端连接
for (int i = 1; i <= server->client_count; i++) {
close(server->clients[i].fd);
}
// 关闭服务器套接字
if (server->server_fd != -1) {
close(server->server_fd);
}
// 释放内存
if (server->clients) {
free(server->clients);
}
printf("服务器资源清理完成\n");
printf("统计信息:\n");
printf(" 总发送消息数: %lu\n", server->messages_sent);
printf(" 总发送字节数: %lu\n", server->bytes_sent);
if (server->messages_sent > 0) {
printf(" 平均消息大小: %.2f 字节\n",
(double)server->bytes_sent / server->messages_sent);
}
}
/**
* 演示高性能网络服务器
*/
int demo_high_performance_server() {
high_perf_server_t server;
printf("=== 高性能网络服务器示例 ===\n");
// 初始化服务器
if (server_init(&server, 8082, 10) != 0) {
return -1;
}
// 启动测试客户端
if (fork() == 0) {
sleep(2); // 等待服务器启动
// 创建多个并发客户端
for (int i = 0; i < 3; i++) {
if (fork() == 0) {
int client_sock = socket(AF_INET, SOCK_STREAM, 0);
struct sockaddr_in serv_addr;
memset(&serv_addr, 0, sizeof(serv_addr));
serv_addr.sin_family = AF_INET;
serv_addr.sin_port = htons(8082);
serv_addr.sin_addr.s_addr = inet_addr("127.0.0.1");
if (connect(client_sock, (struct sockaddr*)&serv_addr, sizeof(serv_addr)) == 0) {
printf("客户端 %d 连接成功\n", i + 1);
// 接收服务器发送的消息
char buffer[256];
for (int msg = 0; msg < 20; msg++) {
ssize_t bytes = recv(client_sock, buffer, sizeof(buffer) - 1, 0);
if (bytes > 0) {
buffer[bytes] = '\0';
printf("客户端 %d 接收消息: %s\n", i + 1, buffer);
} else if (bytes == 0) {
printf("客户端 %d 连接关闭\n", i + 1);
break;
}
}
}
close(client_sock);
exit(0);
}
}
// 等待所有客户端完成
for (int i = 0; i < 3; i++) {
int status;
wait(&status);
}
exit(0);
}
// 运行服务器
server_run(&server);
// 清理资源
server_cleanup(&server);
// 等待测试客户端结束
int status;
wait(&status);
return 0;
}
int main() {
return demo_high_performance_server();
}
|