基于reactor设计websocket服务器

科技 08-15 来源： linux技术栈

1.什么是WebSocket？

WebSocket是HTML5下一种新的协议（websocket协议本质上是一个基于tcp的协议）

websocket使得客户端和服务器之间的数据交换变得更加简单，允许服务端主动向客户端推送数据。在WebSocket API中，浏览器和服务器只需要完成一次握手，两者之间就直接可以创建持久性的连接，并进行双向数据传输。

2.总体过程：

2.1 握手过程：

首先，客户端发起http请求，经过3次握手后，建立起TCP连接；http请求里存放WebSocket支持的版本号等信息，如：Upgrade、Connection、WebSocket-Version等；

然后，服务器收到客户端的握手请求后，同样采用HTTP协议回馈数据；最后，客户端收到连接成功的消息后，开始借助于TCP传输信道进行全双工通信。

第一个一条信息来自|Sec-WebSocket-Key|报头字段在客户端握手中:Sec-WebSocket-Key: fmXyv9eR4PG9L53s09jQLA==对于这个报头字段，服务器必须接受该值(作为当前值)在报头字段中，例如，base64编码的[RFC4648]版本减任何前导和尾随空格)，并将其与全局唯一标识符(GUID， [RFC4122])“258 eafa5-e914-47da -95CA-C5AB0DC85B11”的字符串形式，不太可能被使用网络端点不理解WebSocket协议。一个SHA-1哈希(160位)[FIPS;180-3]，base64编码，然后返回到服务器的握手。

###websocket  浏览器-->服务器GET / HTTP/1.1Host: 192.168.240.128:8888Connection: Upgrade ##升级版Pragma: no-cacheCache-Control: no-cacheUser-Agent: Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/94.0.4606.71 Safari/537.36 Core/1.94.172.400 QQBrowser/11.1.5140.400Upgrade: websocket #websockrtOrigin: nullSec-WebSocket-Version: 13Accept-Encoding: gzip, deflateAccept-Language: zh-CN,zh;q=0.9Sec-WebSocket-Key: fmXyv9eR4PG9L53s09jQLA==Sec-WebSocket-Extensions: permessage-deflate; client_max_window_bits  ###握手信息拼接Key+GUIDKey =  fmXyv9eR4PG9L53s09jQLA==GUID = 258EAFA5-E914-47DA-95CA-C5AB0DC85B11str = Key+GUID = fmXyv9eR4PG9L53s09jQLA==258EAFA5-E914-47DA-95CA-C5AB0DC85B11 ###哈希SHA-1sha = SHA-1(str)###base64转换vaule = base64-encoded(sha); ###服务器-->浏览器HTTP/1.1 101 Switching ProtocolsUpgrade: websocketConnection: UpgradeSec-WebSocket-Accept: vaule

具体代码实现

//响应包拼接int ws_handshark(struct ntyevent *ev) {int idx = 0;char sec_data[128] = {0};char sec_accept[128] = {0}; do{char linebuff[1024] = {0};idx = readline(ev->buffer, idx, linebuff); if(strstr(linebuff, "Sec-WebSocket-Key") > 0){//Sec-WebSocket-Key: hGI6OP19pWseAnpMcEnb2g==//Key+GUID//Sec-WebSocket-Key: hGI6OP19pWseAnpMcEnb2g==258EAFA5-E914-47DA-95CA-C5AB0DC85B11strcat(linebuff, GUID); //哈希SHA-1SHA1(linebuff + strlen("Sec-WebSocket-Key: "), strlen(linebuff + strlen("Sec-WebSocket-Key: ")), sec_data); //base64 编码base64_encode(sec_data, strlen(sec_data), sec_accept); //printf("idx: %d, line: %ld
",idx, sizeof("Sec-WebSocket-Key: "));//printf("idx: %d, line: %ld
",idx, strlen("Sec-WebSocket-Key: ")); printf("idx %d ; line:%s
", idx, sec_accept);memcpy(ev->sec_accept, sec_accept, ACCEPT_KEY_LENGTH);}//printf("line %d ; line:%s
", idx, linebuff);}while((ev->buffer[idx] != '\r' || ev->buffer[idx + 1] != '
') && idx != -1); //两组\r
结束return 0;} //应答 #服务器-->浏览器int ws_response(struct ntyevent* ev){ev->wlength = sprintf(ev->wbuffer, "HTTP/1.1 101 Switching Protocols\r
""Upgrade: websocket\r
""Connection: Upgrade\r
""Sec-WebSocket-Accept: %s\r
\r
", ev->sec_accept); printf("response: %s
", ev->wbuffer);return ev->wlength;}

相关视频推荐

深入理解websocket，为你的项目多条思路

linux多线程之epoll原理剖析与reactor原理及应用

学习地址：C/C++Linux服务器开发/后台架构师【零声教育】-学习视频教程-腾讯课堂

需要C/C++ Linux服务器架构师学习资料加qun812855908获取（资料包括C/C++，Linux，golang技术，Nginx，ZeroMQ，MySQL，Redis，fastdfs，MongoDB，ZK，流媒体，CDN，P2P，K8S，Docker，TCP/IP，协程，DPDK，ffmpeg等），免费分享

2.2 接收和发送

websocket协议的数据帧

FIN :1bit
指示这是消息中的最后一个片段。第一个碎片也可能是最后的碎片。
RSV1、RSV2、RSV3:各1bit
必须为0，除非协商一个定义含义的扩展为非零值。如果接收到一个非零值，并且没有
协商好的扩展定义了这种非零的含义值，接收端点必须失败WebSocketConnection.
Opcode: 4 bits
定义“有效负载数据”的解释。如果一个未知的操作码，接收端点必须失败WebSocket Connection_。定义了以下值。
* %x0 表示延续帧
* %x1 表示文本框架
* %x2 表示二进制帧
* %x3-7 为其他非控制帧保留
* %x8 表示连接关闭
* %x9 表示ping
* %xA 表示pong
* %xB-F 为进一步的控制帧保留
Mask: 1 bit
定义是否屏蔽“有效负载数据”。如果设置为1，则a屏蔽键包含在屏蔽键中，用于解除屏蔽。
解除屏蔽
j = i MOD 4
transformed-octet-i = original-octet-i XOR masking-key-octet-j
Payload length: 7 bits, 7+16 bits, or 7+64 bits
“有效载荷数据”的长度，以字节为单位:如果0-125，则为有效载荷长度。如果126，下面的2个字节被解释为a16位无符号整数是有效载荷长度。如果127,后面的8个字节被解释为64位无符号整数最有效位必须为0)为有效载荷长度。多字节长度量用网络字节顺序表示。请注意,在所有情况下，必须使用最小字节数进行编码长度。

将数据帧映射到结构体，小端字节序

struct ws_ophdr {  //小端 unsigned char opcode:4,rsv3:1,rsv2:1,rsv1:1,fin:1; unsigned char pl_len:7,   mask:1; };

具体代码

//Masking-key, if MASK set to 1 解除数据屏蔽void umask(char *payload, int length, char *mask_key) { int i = 0; for (i = 0;i < length;i ++) {payload[i] ^= mask_key[i%4];} } int ws_tranmission(struct ntyevent *ev) {struct ws_ophdr *hdr = (struct ws_ophdr *)ev->buffer; if (hdr->pl_len < 126) { unsigned char *payload = NULL;if (hdr->mask) {payload = ev->buffer + 6; umask(payload, hdr->pl_len, ev->buffer + 2);} else {payload = ev->buffer + 2;} printf("payload: %s
", payload); } else if (hdr->pl_len == 126) { } else if (hdr->pl_len == 127)  { } else {//assert(0);} return 0;}

3. 完整代码

//gcc reactor_server_websocket.c -o server  -lssl -lcrypto#include #include #include #include #include #include #include #include #include  #include #include #include #include   #define GUID  "258EAFA5-E914-47DA-95CA-C5AB0DC85B11" //状态机enum {WS_HANDSHARK = 0,    //握手状态WS_TRANMISSION = 1,//数据传输WS_END = 2,//终止传输WS_COUNT};  struct ws_ophdr {  //小端 unsigned char opcode:4,rsv3:1,rsv2:1,rsv1:1,fin:1; unsigned char pl_len:7,   mask:1; };  #define  BUFFER_LENGTH  1024#define  ACCEPT_KEY_LENGTH64 #define  MAX_EPOLL_EVENTS   1024  //epoll事件数量#define  SERVER_PORT  8888#define  PORT_COUNT1 typedef int NCALLBACK(int ,int ,void*); //管理每一个io fd的结构体struct ntyevent{int fd; //io fdint events;void *arg;int (*callback)(int fd, int events, void* arg); //执行回调函数 int status;//判断是否已有事件char buffer[BUFFER_LENGTH]; //用户缓冲区 //requestint length; //用户缓冲区长度 char wbuffer[BUFFER_LENGTH]; //responseint wlength;char sec_accept[ACCEPT_KEY_LENGTH]; int wsstatus; //0, 1, 2, 3 描述状态机}; //管理ntyevent fd的块struct eventblock{struct eventblock* next; //指向ntyevent fd集合struct ntyevent* events; //指向下一个ntyevent fd的块}; //reacotr结点struct ntyreactor{int epfd;//epoll fdint blkcnt;//ntyevent fd的块 计数 struct eventblock* evblks;//指向ntyevent fd的块头结点}; int recv_cb(int fd, int events, void *arg);int send_cb(int fd, int events, void *arg);int accept_cb(int fd, int events, void* arg); struct ntyevent *ntyreactor_idx(struct ntyreactor *reactor, int sockfd); //io fd结构体设置void nty_event_set(struct ntyevent* ev, int fd, NCALLBACK callback, void* arg){ev->fd = fd;ev->callback = callback;ev->events = 0;ev->arg = arg; return ;} //io fd addint nty_event_add(int epfd, int events, struct ntyevent *ev){struct epoll_event ep_ev = {0, {0}};ep_ev.data.ptr = ev;//io fd结构体ep_ev.events = ev->events = events; //需要检测的fd事件 int op; //操作类型if(ev->status == 1){op = EPOLL_CTL_MOD; //修改}else{op = EPOLL_CTL_ADD; //添加ev->status = 1;//标志已经添加}if(epoll_ctl(epfd, op, ev->fd, &ep_ev) <0 ){ //绑定printf("event add failed [fd=%d], events[%d]
", ev->fd, events);return -1;} return 0;} //io fd delint nty_event_del(int epfd, struct ntyevent* ev){struct epoll_event ep_ev = {0, {0}};if(ev->status != 1){ //没有添加过检测的fd事件return -1;} ep_ev.data.ptr = ev;ev->status = 0; //标志未添加epoll_ctl(epfd, EPOLL_CTL_DEL, ev->fd, &ep_ev);return 0;} //读行int readline(char* allbuf, int idx, char* linebuf){int len = strlen(allbuf); for(; idx < len; idx++){if(allbuf[idx] == '\r' && allbuf[idx+1] == '
')return idx+2;else*(linebuf++) = allbuf[idx];}return -1;} int base64_encode(char *in_str, int in_len, char *out_str) {    BIO *b64, *bio;    BUF_MEM *bptr = NULL;    size_t size = 0;     if (in_str == NULL || out_str == NULL)        return -1;     b64 = BIO_new(BIO_f_base64());    bio = BIO_new(BIO_s_mem());    bio = BIO_push(b64, bio);BIO_write(bio, in_str, in_len);    BIO_flush(bio);     BIO_get_mem_ptr(bio, &bptr);    memcpy(out_str, bptr->data, bptr->length);    out_str[bptr->length-1] = '\0';    size = bptr->length;     BIO_free_all(bio);    return size;} //握手流程int ws_handshark(struct ntyevent *ev) {int idx = 0;char sec_data[128] = {0};char sec_accept[128] = {0}; do{char linebuff[1024] = {0};idx = readline(ev->buffer, idx, linebuff); if(strstr(linebuff, "Sec-WebSocket-Key") > 0){//Sec-WebSocket-Key: hGI6OP19pWseAnpMcEnb2g==//Key+GUID//Sec-WebSocket-Key: hGI6OP19pWseAnpMcEnb2g==258EAFA5-E914-47DA-95CA-C5AB0DC85B11strcat(linebuff, GUID); //哈希SHA-1SHA1(linebuff + strlen("Sec-WebSocket-Key: "), strlen(linebuff + strlen("Sec-WebSocket-Key: ")), sec_data); //base64 编码base64_encode(sec_data, strlen(sec_data), sec_accept); //printf("idx: %d, line: %ld
",idx, sizeof("Sec-WebSocket-Key: "));//printf("idx: %d, line: %ld
",idx, strlen("Sec-WebSocket-Key: ")); printf("idx %d ; line:%s
", idx, sec_accept);memcpy(ev->sec_accept, sec_accept, ACCEPT_KEY_LENGTH);}//printf("line %d ; line:%s
", idx, linebuff);}while((ev->buffer[idx] != '\r' || ev->buffer[idx + 1] != '
') && idx != -1); //两组\r
结束return 0;} //Masking-key, if MASK set to 1 解除数据屏蔽void umask(char *payload, int length, char *mask_key) { int i = 0; for (i = 0;i < length;i ++) {payload[i] ^= mask_key[i%4];} } int ws_tranmission(struct ntyevent *ev) {struct ws_ophdr *hdr = (struct ws_ophdr *)ev->buffer; if (hdr->pl_len < 126) { unsigned char *payload = NULL;if (hdr->mask) {payload = ev->buffer + 6; umask(payload, hdr->pl_len, ev->buffer + 2);} else {payload = ev->buffer + 2;} printf("payload: %s
", payload); } else if (hdr->pl_len == 126) { } else if (hdr->pl_len == 127)  { } else {//assert(0);} return 0;} //请求 协议解析int ws_request(struct ntyevent * ev){if (ev->wsstatus == WS_HANDSHARK) {ws_handshark(ev);ev->wsstatus = WS_TRANMISSION;  //状态改变} else if (ev->wsstatus == WS_TRANMISSION) {ws_tranmission(ev);}}   //应答 #服务器-->浏览器int ws_response(struct ntyevent* ev){ev->wlength = sprintf(ev->wbuffer, "HTTP/1.1 101 Switching Protocols\r
""Upgrade: websocket\r
""Connection: Upgrade\r
""Sec-WebSocket-Accept: %s\r
\r
", ev->sec_accept); printf("response: %s
", ev->wbuffer);return ev->wlength;}   //recv回调int recv_cb(int fd, int events, void* arg){struct ntyreactor* reactor = (struct ntyreactor*)arg;struct ntyevent* ev = ntyreactor_idx(reactor, fd); if(ev == NULL)return -1; int len = recv(fd, ev->buffer, BUFFER_LENGTH, 0);nty_event_del(reactor->epfd, ev); if (len > 0) {ev->length = len;ev->buffer[len] = '\0'; //printf("recv [%d]:%s
", fd, ev->buffer); ws_request(ev); //将fd 设置为发送事件nty_event_set(ev, fd, send_cb, reactor);nty_event_add(reactor->epfd, EPOLLOUT, ev);} else if (len == 0) {  //客户端断开连接 nty_event_del(reactor->epfd, ev);printf("recv_cb --> disconnect
");close(ev->fd); } else { //返回错误 if (errno == EAGAIN && errno == EWOULDBLOCK) { //} else if (errno == ECONNRESET){nty_event_del(reactor->epfd, ev);close(ev->fd);}printf("recv[fd=%d] error[%d]:%s
", fd, errno, strerror(errno));} return len;}   //send回调int send_cb(int fd, int events, void* arg){struct ntyreactor* reactor = (struct ntyreactor*)arg;struct ntyevent* ev = ntyreactor_idx(reactor, fd); if (ev == NULL) return -1; ws_response(ev); int len = send(fd, ev->wbuffer, ev->length, 0);if (len > 0) {printf("send[fd=%d], [%d]%s
", fd, len, ev->wbuffer); //发送后，将fd设置为接收事件nty_event_del(reactor->epfd, ev);nty_event_set(ev, fd, recv_cb, reactor);nty_event_add(reactor->epfd, EPOLLIN, ev);} else { //发送失败 nty_event_del(reactor->epfd, ev);close(ev->fd); printf("send[fd=%d] error %s
", fd, strerror(errno)); } return len;}  //客户端接入回调int accept_cb(int fd, int events, void* arg){struct ntyreactor *reactor = (struct ntyreactor*)arg;if (reactor == NULL) return -1; struct sockaddr_in client_addr;socklen_t len = sizeof(client_addr); int clientfd; //客户端接入if ((clientfd = accept(fd, (struct sockaddr*)&client_addr, &len)) == -1) {if (errno != EAGAIN && errno != EINTR) {}printf("accept: %s
", strerror(errno));return -1;} //设置非阻塞fdint flag = 0;if ((flag = fcntl(clientfd, F_SETFL, O_NONBLOCK)) < 0) {printf("%s: fcntl nonblocking failed, %d
", __func__, MAX_EPOLL_EVENTS);return -1;} struct ntyevent *event = ntyreactor_idx(reactor, clientfd); if (event == NULL) return -1; //将该fd设置为recvnty_event_set(event, clientfd, recv_cb, reactor);nty_event_add(reactor->epfd, EPOLLIN, event); printf("new connect [%s:%d], pos[%d]
", inet_ntoa(client_addr.sin_addr), ntohs(client_addr.sin_port), clientfd); return 0;} //创建socket监听int init_sock(short port){int fd = socket(AF_INET, SOCK_STREAM, 0);fcntl(fd, F_SETFL, O_NONBLOCK); struct sockaddr_in server_addr;memset(&server_addr, 0, sizeof(server_addr));server_addr.sin_family = AF_INET;server_addr.sin_addr.s_addr = htonl(INADDR_ANY);server_addr.sin_port = htons(port); bind(fd, (struct sockaddr*)&server_addr, sizeof(server_addr)); if (listen(fd, 20) < 0) {printf("listen failed : %s
", strerror(errno));return -1;} printf("listen server port : %d
", port);return fd;}  //reactor扩展大小int ntyreactor_alloc(struct ntyreactor* reactor){if(reactor == NULL) return -1;if(reactor->evblks == NULL) return -1; struct eventblock* blk = reactor->evblks; //块的头结点 //找尾节点while(blk->next != NULL){  //找到尾节点blk = blk->next;} struct ntyevent* evs = (struct ntyevent*)malloc((MAX_EPOLL_EVENTS) * sizeof(struct ntyevent));if (evs == NULL) {printf("ntyreactor_alloc ntyevent failed
");return -2;}memset(evs, 0, (MAX_EPOLL_EVENTS) * sizeof(struct ntyevent)); struct eventblock *block = malloc(sizeof(struct eventblock));if (block == NULL) {printf("ntyreactor_alloc eventblock failed
");return -3;}//io fd集合连接成块block->events = evs;block->next = NULL; //指向新块blk->next = block;reactor->blkcnt ++; return 0;} //根据io fd来找fd结构体struct ntyevent *ntyreactor_idx(struct ntyreactor *reactor, int sockfd) {if (reactor == NULL) return NULL;if (reactor->evblks == NULL) return NULL; int blkidx = sockfd / MAX_EPOLL_EVENTS; //在哪一个块while (blkidx >= reactor->blkcnt) {//大小不够扩容ntyreactor_alloc(reactor);} int i = 0;struct eventblock *blk = reactor->evblks; //头结点块while (i++ != blkidx && blk != NULL) {    //找到所在的块blk = blk->next;} return &blk->events[sockfd % MAX_EPOLL_EVENTS]; //返回fd结构体} //reactor初始化int ntyreactor_init(struct ntyreactor* reactor){if(reactor == NULL) return -1;memset(reactor, 0, sizeof(struct ntyreactor)); reactor->epfd = epoll_create(1); if (reactor->epfd <= 0) {printf("create epfd in %s err %s
", __func__, strerror(errno));return -2;} //创建第一个块struct ntyevent* evs = (struct ntyevent*)malloc((MAX_EPOLL_EVENTS) * sizeof(struct ntyevent));if (evs == NULL) {printf("create epfd in %s err %s
", __func__, strerror(errno));close(reactor->epfd);return -3;}memset(evs, 0, (MAX_EPOLL_EVENTS) * sizeof(struct ntyevent)); struct eventblock *block = malloc(sizeof(struct eventblock));if (block == NULL) {free(evs);close(reactor->epfd);return -3;}block->events = evs;block->next = NULL; reactor->evblks = block;reactor->blkcnt = 1; return 0;} //销毁reactorint ntyreactor_destory(struct ntyreactor* reactor){close(reactor->epfd); struct eventblock *blk = reactor->evblks;struct eventblock *blk_next;while (blk != NULL) {blk_next = blk->next; free(blk->events);free(blk);blk = blk_next;} return 0;} //初始化接收连接socketint ntyreactor_addlistener(struct ntyreactor* reactor, int sockfd, NCALLBACK *acceptor){if (reactor == NULL) return -1;if (reactor->evblks == NULL) return -1; struct ntyevent* event = ntyreactor_idx(reactor, sockfd);if (event == NULL) return -1; nty_event_set(event, sockfd, acceptor, reactor);nty_event_add(reactor->epfd, EPOLLIN, event); return 0;} //reactor事件循环int ntyreactor_run(struct ntyreactor* reactor){if (reactor == NULL) return -1;if (reactor->epfd < 0) return -1;if (reactor->evblks == NULL) return -1;struct epoll_event events[MAX_EPOLL_EVENTS+1];int checkpos = 0, i; while(1){int nready = epoll_wait(reactor->epfd, events, MAX_EPOLL_EVENTS, 1000);if (nready < 0) {printf("epoll_wait error, exit
");continue;} for(i = 0;i < nready; i++){struct ntyevent* ev = (struct ntyevent*)events[i].data.ptr; //发生事件的io fd结构体 if((events[i].events & EPOLLIN) && (ev->events & EPOLLIN)){ev->callback(ev->fd, events[i].events, ev->arg);}if((events[i].events & EPOLLOUT) && (ev->events & EPOLLOUT)){ev->callback(ev->fd, events[i].events, ev->arg);}} } } int main(int argc, char *argv[]) { struct ntyreactor *reactor = (struct ntyreactor*)malloc(sizeof(struct ntyreactor));ntyreactor_init(reactor); //起始的端口号unsigned short port = SERVER_PORT;if (argc == 2) {port = atoi(argv[1]);} int i = 0;int sockfds[PORT_COUNT] = {0}; for (i = 0;i < PORT_COUNT;i ++) {sockfds[i] = init_sock(port+i);ntyreactor_addlistener(reactor, sockfds[i], accept_cb);} ntyreactor_run(reactor); ntyreactor_destory(reactor);for (i = 0;i < PORT_COUNT;i ++) {close(sockfds[i]);}free(reactor); return 0;}