江苏龙网

  1. 主页 > 生活百科 > >

浅谈linux下基于UDP服务的负载均衡方法( 三 )

负载均衡


可是要充分利用多CPU性能 ， 最重要的还是要使用多线程/多进程结构 ， 特别是在处理UDP“长连接”的时候 ， 说起UDP长连接 ， 可能有人会觉得我把概念弄糊涂了 ， 实际上我清楚得很 ， 只不过没有更好的词也不想发明新的词来形容使用UDP协议时的OpenVPN是怎么收发数据的 。至于短连接且不太频繁的 ， 多进程意义倒不是很大 ， 反而会引入进程创建 ， 进程切换带来的开销 。那么多进程自然是好的 ， 如何实现呢？自然而然想到的就是预先建立多个进程了 ， 每一个进程分配一个UDP socket来处理数据的收发 ， 涉及到细节有很多种：
a.写一个程序 ， 创建一个使用SO_REUSEPORT绑定一个IP和端口的UDP socket ， 然后启动多个实例 ， 启动脚本可以独立于程序以外；
b.在主进程中预先fork多个进程 ， 每一个进程只处理一个UDP socket；
c.在主进程中预先fork多个进程 ， 每一个进程可以处理全部的UDP socket；
d.试图建立UDP的accept模型
以上的方案中 ， 第一个很明显没有什么意思 ， 它也是我着手OpenVPN负载均衡的第一个方案 ， 很傻瓜 ， 很简单 ， 很实用 ， 没什么技巧可以炫耀 ， 没什么“技术含量” ， 以前的文章中说的多了 ， 本文就不说了 ， 只不过之前不能绑定相同端口(其实可以 ， 但是绑了也没有用) 。另外的3个方案中 ， 每一个都可以讲一段故事 。在讲故事之前 ， 首先给出方案b的代码 ， 因为它最纯粹 ， 也最成功 ， 也是唯一可行的方案 。代码如下：
#include <sys/types.h>#include <string.h>#include <netinet/in.h>#include <stdio.h>#include <stdlib.h>#include <string.h>#include <sys/socket.h>#include <unistd.h>#include <sys/wait.h>#include <fcntl.h>#include <sys/uio.h>#define NUM_PROCESS10#define SO_REUSEPORT15#define SERV_PORT12345#define MAXSIZE1024struct worker_arg {int sock_fd;int process_num;};void DEBUG(char *argv) {//TODO}void client_echo(int sock, int process_num){int n;char buff[MAXSIZE];struct sockaddr clientfrom;socklen_t addrlen = sizeof(clientfrom);for(;;) {memset(buff, 0, MAXSIZE);n = recvfrom(sock, buff, MAXSIZE, 0, &clientfrom, &addrlen);printf("%dn", process_num);n = sendto(sock, buff, n, 0, &clientfrom, addrlen);}}int create_udp_sock(const char *str_addr){int sockfd;int optval = 1;int fdval = 0;struct sockaddr_in servaddr, cliaddr;sockfd = socket(AF_INET, SOCK_DGRAM|SOCK_CLOEXEC, 0); /* create a socket */setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, &optval, sizeof(optval));//这是关键 ， 关键中的关键 。如果没有这个 ， 此代码在之前内核照样完美运行！完美？完美！setsockopt(sockfd, SOL_SOCKET, SO_REUSEPORT, &optval, sizeof(optval));bzero(&servaddr, sizeof(servaddr));servaddr.sin_family = AF_INET;servaddr.sin_addr.s_addr = inet_addr(str_addr);servaddr.sin_port = htons(SERV_PORT);if(bind(sockfd, (struct sockaddr *)&servaddr, sizeof(servaddr)) == -1) {perror("bind error");exit(1);}perror("bind ");return sockfd;}void woker_process(int sock, int num){int woker_sock = sock;//recv_fd(sock);if(woker_sock < 0){}client_echo(woker_sock, num);}pid_t create_work(void *arg){struct worker_arg *warg = (struct worker_arg*)arg;pid_t pid = fork();if (pid > 0) {return pid;} else if (pid == 0){woker_process(warg->sock_fd, warg->process_num);// 不可能运行到此} else {exit(1);}// 不可能运行到此return pid;}void schedule_process(int socks[2], char *str_addr){pid_t pids[NUM_PROCESS] = {0};int udps[NUM_PROCESS] = {0};int i = 0;for (i = 0; i < NUM_PROCESS; i++) {udps[i] = create_udp_sock(str_addr);}for (i = 0; i < NUM_PROCESS; i++) {struct worker_arg warg;warg.sock_fd = udps[i];warg.process_num = i;pid_t pid = create_work(&warg);pids[i] = pid;}while (1) {int stat;int i;pid_t pid = waitpid(-1, &stat, 0);for (i = 0; i < NUM_PROCESS; i++) {if (pid == pids[i]) {// 最最关键的 ， 那就是”把特定的套接字传递到特定的进程中“struct worker_arg warg;warg.sock_fd = udps[i];warg.process_num = i;pid_t pid = create_work(&warg);pids[i] = pid;break;}}}}int main(int argc, char** argv){int unix_socks[2];char *str_addr = argv[1];if(socketpair(AF_UNIX, SOCK_STREAM, 0, unix_socks) == -1){exit(1);}schedule_process(unix_socks, str_addr);return 0;}最最重要的是上面的那个wait逻辑 ， 你要保证 ， 一个进程挂掉以后 ， 要将其迅速拉起来 ， 因为如果不这样 ， 所有命中(通过第四层的socket查找算法)该socket的数据都将得不到处理 。


推荐阅读


上一篇：用 Python 开发一个 「聊天室」

下一篇：两种证件照更换背景色的方法 简单又实用