| 编辑推荐: |
本篇文章主要介绍websocket和http的瓜葛，websocket协议的实现分块分析, 如何在reactor的基础上封装websocket应用层协议 (哪些协议究竟是如何封装实现的)，详见参考文章。
本文来自CSDN的博客，由火龙果软件Alice编辑推荐。 |
|
websocket和http的瓜葛
http的弊端引出为什么需要websocket
http是一种无状态, 无连接, 非持久化 的单向半双工应用层协议
啥叫作无状态, 对于历史连接是完全没有记忆的, 每一次连接都是新的连接
无连接的和非持久化其实是一个意思, 一次请求, 一次响应, 不会持续.
单向半双工指的是 通信请求只能由客户端发起, 服务端只能对于请求做出应答, 服务端不能主动地向客户端发送数据
引出问题 --- 服务器无法主动向客户端发送数据, 如果服务端存在一定地状态变更, 却无法实时地主动向客户端推送这个数据
针对上述问题, 最开始地时候还是使用http, 只不过通过定时轮询和长轮询地方式来解决服务器需要向客户端主动发送数据地问题.
定时轮询， 不断地定时地询问服务器, 你需要发送消息吗, 不断地请求, 定时发出询问, 如果服务器需要发送数据到客户端, 询问来了就可以发送状态变更，数据到客户端了
定时轮询地弊端：可能存在很严重地延时性， 而且不断地轮询及其浪费占用服务端地资源, 增大服务端压力,不断地建立连接浪费资源 存在很多无效请求 服务器表示，不停的建立连接，大量消耗我的带宽和资源, 我需要很快的处理连接 , 而很多时候我都没有数据跟新, 存在大量无效请求 （服务器表示我很被动呀）
借鉴了大佬写的前端的websocket实现一个网页聊天室, 而且对于websocket的解释也是相当的优秀,
我们平时点外卖 (轮询实例)
0秒：食物到了吗？（客户）
0秒：正在配送。（外卖小哥）
1秒：食物到了吗？（客户）
1秒：正在配送。(外卖小哥)
2秒：食物到了吗？（客户）
2秒：正在配送。（外卖小哥） 2秒及其之间那么多询问都是无效询问
3秒：食物到了吗？(客户)
3秒：是的，先生，这是您的外卖。（外卖小哥）
升级为长轮询，也仅仅只能解决延时性地问题, 能达到实时地将服务端地状态数据推送到客户端地目的, 但是http连接始终打开，长连接，浪费系统资源, 客户端需要等待服务端响应，服务端也一直被客户端占用. (一直占用服务器, 无效占用)
长轮询生活实例
0秒：食物到了吗？（客户）
。。。 中间电话一直不挂断, 直到外卖送到
3秒：是的，先生，这是您的外卖。（外卖小哥）
解决问题 --- websocket全双工地通讯协议地诞生, 服务器可以主动向客户端发送数据
阶段分为 一次握手阶段 + 数据交互阶段
主要核心：全双工, 服务器可以主动向客户端发送数据
websocket的特点
建立在TCP协议上, 服务器端的实现比较容易
与HTTP协议有着良好的兼容性, 默认端口也是80和443，并且握手阶段基于HTTP协议
数据格式比较轻量，性能开销小，通信高效
可以发送文本, 也可以发送二进制数据
报文分析
GET /chat HTTP/1.1 #请求行
Host: server.example.com
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==
Origin: http://example.com
Sec-WebSocket-Protocol: chat, superchat
Sec-WebSocket-Version: 13
#请求头部 |
注意: 上述框着的升级为websocket其实是告知服务器客户端想要建立的是websocket连接, 你支持吗, 如果服务器支持，在响应报文中也一定存在返回这两个头部字段
响应
HTTP/1.1 101 Switching Protocols #响应行,状态行
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=
Sec-WebSocket-Protocol: chat
# 响应报头 |
Sec-WebSocket-Accept其实是根据客户端的Key做Hash之后返回的结果
websocket在我们生活中的实例场景(服务器(后端)向网页客户端(前端)实时刷新数据)
1.弹幕的实时刷新
2.扫描微信二维码后的页面跳转
3.股票数据的实时刷新
websocket协议的实现分块分析, 如何在reactor的基础上封装websocket应用层协议 (哪些协议究竟是如何封装实现的)
过程分析
握手细节:
基于TCP连接完成之后，进行一次握手的意义
握手：确保服务器是支持websocket 协议的, 也就是服务器应答一下客户端, 你的升级请求我收到了，我是支持websocket升级的
细节分析: 如何区别握手数据 和 普通交互数据 ?
其实核心在于区分不同的阶段, 状态 --- 状态机, 区分各个状态
状态机 --- http协议底层也存在， 协议封装必不可少的部分，因为需要进行区分不同的阶段, 是握手建立连接的阶段, 还是数据交互的阶段 ... 这些http协议底层肯定是需要区分的
每一个连接都拥有如下三种状态
enum WEBSOCKET_STATUS {
WS_HANDSHARK,//握手状态
WS_DATATRANSFORM,//数据交互状态
WS_DATAEND,//断开状态
}; |
握手细节核心: Sec-WebSocket-Key ---> Sec-WebSocket-Accept
为什么 需要经过层层加密 key -> Accept key ？
为了安全起见, 为了证明它们可以处理websocket请求。 密匙认证.
加密过程伪代码如下: 获取accept.
//伪代码如下
str = Sec-WebSocket-Key;
//拿出Sec-WebSocket-Key客户端序列串
str += GUID;
sha = SHA-1(str); //SHA-1一种hash
accept = base64_encode(sha); |
transform 数据推送的细节 --- 数据封包和解包.
如下是websocket独有的数据帧格式, 握手之后的数据发送都需要按照数据帧的格式对需要发送的数据进行一个处理
做自定义协议必须的三部分. 基于tcp的自定义应用层协议
1.操作码. eg: FIN RSV
2.包长度
3.mask-key
数据封包函数 + 数据解包函数 (具体的封包解包过程我还没有吃透, 如果后面有机会小杰希望可以重新分析来过, 如下目前是借鉴的前辈的代码)
void umask(char *data,int len,char *mask) {
int i;
for (i = 0;i < len;i ++)
*(data+i) ^= *(mask+(i%4));
}
char* decode_packet(char *stream, char *mask, int length, int *ret) {
nty_ophdr *hdr = (nty_ophdr*)stream;
unsigned char *data = stream + sizeof(nty_ophdr);
int size = 0;
int start = 0;
//char mask[4] = {0};
int i = 0;
//if (hdr->fin == 1) return NULL;
if ((hdr->mask & 0x7F) == 126) {
nty_websocket_head_126 *hdr126 = (nty_websocket_head_126*)data;
size = hdr126->payload_length;
for (i = 0;i < 4;i ++) {
mask[i] = hdr126->mask_key[i];
}
start = 8;
} else if ((hdr->mask & 0x7F) == 127) {
nty_websocket_head_127 *hdr127 = (nty_websocket_head_127*)data;
size = hdr127->payload_length;
for (i = 0;i < 4;i ++) {
mask[i] = hdr127->mask_key[i];
}
start = 14;
} else {
size = hdr->payload_length;
memcpy(mask, data, 4);
start = 6;
}
*ret = size;
umask(stream+start, size, mask);
return stream + start;
}
int encode_packet(char *buffer,char *mask, char *stream, int length) {
nty_ophdr head = {0};
head.fin = 1;
head.opcode = 1;
int size = 0;
if (length < 126) {
head.payload_length = length;
memcpy(buffer, &head, sizeof(nty_ophdr));
size = 2;
} else if (length < 0xffff) {
nty_websocket_head_126 hdr = {0};
hdr.payload_length = length;
memcpy(hdr.mask_key, mask, 4);
memcpy(buffer, &head, sizeof(nty_ophdr));
memcpy(buffer+sizeof(nty_ophdr), &hdr, sizeof(nty_websocket_head_126));
size = sizeof(nty_websocket_head_126);
} else {
nty_websocket_head_127 hdr = {0};
hdr.payload_length = length;
memcpy(hdr.mask_key, mask, 4);
memcpy(buffer, &head, sizeof(nty_ophdr));
memcpy(buffer+sizeof(nty_ophdr), &hdr, sizeof(nty_websocket_head_127));
size = sizeof(nty_websocket_head_127);
}
memcpy(buffer+2, stream, length);
return length + 2;
} |
再注意一下recv数据之后需要按照不同的状态进行调用不同的函数进行处理数据
整体代码:
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <netinet/tcp.h>
#include <arpa/inet.h>
#include <pthread.h>
#include <fcntl.h>
#include <errno.h>
#include <sys/epoll.h>
#include <openssl/sha.h>
#include <openssl/pem.h>
#include <openssl/bio.h>
#include <openssl/evp.h>
typedef struct sockaddr SA;
#define BUFFSIZE 1024
#define GUID "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"
enum WEBSOCKET_STATUS {
WS_HANDSHARK,//握手状态
WS_DATATRANSFORM,//数据交互状态
WS_DATAEND,//断开状态
};
struct sockitem {
int sockfd;
int (*callback)(int fd, int events, void* arg);
//arg 传入 sockitem*
char recvbuffer[BUFFSIZE];
char sendbuffer[BUFFSIZE];
int rlen;//recvlen
int slen;//sendlen
int status;//存储状态
};
//mainloop / eventloop
struct reactor {
int epfd;
struct epoll_event events[512];
};
struct reactor* eventloop = NULL; //事件循环
int recv_cb(int fd, int events, void* arg);
int send_cb(int fd, int events, void* arg);
// websocket
char* decode_packet(char *stream, char *mask, int length, int *ret);
int encode_packet(char *buffer,char *mask, char *stream, int length);
struct _nty_ophdr {
unsigned char opcode:4,
rsv3:1,
rsv2:1,
rsv1:1,
fin:1;
unsigned char payload_length:7,
mask:1;
} __attribute__ ((packed));
struct _nty_websocket_head_126 {
unsigned short payload_length;
char mask_key[4];
unsigned char data[8];
} __attribute__ ((packed));
struct _nty_websocket_head_127 {
unsigned long long payload_length;
char mask_key[4];
unsigned char data[8];
} __attribute__ ((packed));
typedef struct _nty_websocket_head_127 nty_websocket_head_127;
typedef struct _nty_websocket_head_126 nty_websocket_head_126;
typedef struct _nty_ophdr nty_ophdr;
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;
}
//读取一行, allbuff整个缓冲区, level 当前ind linebuff 存储一行
int readline(char* allbuff, int level, char* linebuff ) {
int n = strlen(allbuff);
for (; level < n; ++level) {
//\r\n 回车换行, 表示行末
if (allbuff[level] == '\r' && allbuff[level + 1] == '\n') {
return level + 2;
} else {
*(linebuff++) = allbuff[level]; //存储行数据
}
}
return -1;
}
//握手,
int handshark(struct sockitem* si, struct reactor* mainloop) {
char linebuff[256];//存储一行
char sec_accept[32];//存储进行处理之后的子序列
unsigned char sha1_data[SHA_DIGEST_LENGTH + 1] = {0};
char head[BUFFSIZE] = {0};//存储整个头部信息
int level = 0;
//读取Sec-WebSocket-Key并且处理获取accept-key返回密匙
do {
memset(linebuff, 0, sizeof(linebuff));//清空
level = readline(si->recvbuffer, level, linebuff);
if (strstr(linebuff, "Sec-WebSocket-Key") != NULL) {
//说明是key 值, 需要进行加密处理
strcat(linebuff, GUID);//str += GDID
SHA1((unsigned char*)&linebuff + 19, strlen(linebuff + 19), (unsigned char*)&sha1_data);
//SHA1(str);
base64_encode(sha1_data, strlen(sha1_data), sec_accept);
//将数据全部放入到head中, 写响应信息
sprintf(head, "HTTP/1.1 101 Switching Protocols\r\n"\
"Upgrade: websocket\r\n" \
"Connection: Upgrade\r\n" \
"Sec-WebSocket-Accept: %s\r\n" \
"\r\n", sec_accept);
printf("response\n");
printf("%s\n\n\n", head);
//然后进行将其加入到reactor中
memset(si->recvbuffer, 0, BUFFSIZE);
memcpy(si->sendbuffer, head, strlen(head));//to send
si->slen = strlen(head);
//to set epollout events;
struct epoll_event ev;
ev.events = EPOLLOUT | EPOLLET;
//si->sockfd = si->sockfd;
si->callback = send_cb;
//握手完成 --》 状态数据交互
si->status = WS_DATATRANSFORM;
ev.data.ptr = si;
epoll_ctl(mainloop->epfd, EPOLL_CTL_MOD, si->sockfd, &ev);
//握手之后接下来server 需要关注send数据
break;
}
} while ((si->recvbuffer[level] != '\r' || si->recvbuffer[level + 1] != '\n') && level != -1);
return 0;
}
//数据交互函数
int transform(struct sockitem *si, struct reactor *mainloop) {
int ret = 0;
char mask[4] = {0};
char *data = decode_packet(si->recvbuffer, mask, si->rlen, &ret);
printf("data : %s , length : %d\n", data, ret);
ret = encode_packet(si->sendbuffer, mask, data, ret);
si->slen = ret;
memset(si->recvbuffer, 0, BUFFSIZE);
struct epoll_event ev;
ev.events = EPOLLOUT | EPOLLET;
//ev.data.fd = clientfd;
si->sockfd = si->sockfd;
si->callback = send_cb;
si->status = WS_DATATRANSFORM;//标识IO事件处于数据交互状态.
ev.data.ptr = si;
epoll_ctl(mainloop->epfd, EPOLL_CTL_MOD, si->sockfd, &ev);
return 0;
}
void umask(char *data,int len,char *mask) {
int i;
for (i = 0;i < len;i ++)
*(data+i) ^= *(mask+(i%4));
}
char* decode_packet(char *stream, char *mask, int length, int *ret) {
nty_ophdr *hdr = (nty_ophdr*)stream;
unsigned char *data = stream + sizeof(nty_ophdr);
int size = 0;
int start = 0;
//char mask[4] = {0};
int i = 0;
//if (hdr->fin == 1) return NULL;
if ((hdr->mask & 0x7F) == 126) {
nty_websocket_head_126 *hdr126 = (nty_websocket_head_126*)data;
size = hdr126->payload_length;
for (i = 0;i < 4;i ++) {
mask[i] = hdr126->mask_key[i];
}
start = 8;
} else if ((hdr->mask & 0x7F) == 127) {
nty_websocket_head_127 *hdr127 = (nty_websocket_head_127*)data;
size = hdr127->payload_length;
for (i = 0;i < 4;i ++) {
mask[i] = hdr127->mask_key[i];
}
start = 14;
} else {
size = hdr->payload_length;
memcpy(mask, data, 4);
start = 6;
}
*ret = size;
umask(stream+start, size, mask);
return stream + start;
}
int encode_packet(char *buffer,char *mask, char *stream, int length) {
nty_ophdr head = {0};
head.fin = 1;
head.opcode = 1;
int size = 0;
if (length < 126) {
head.payload_length = length;
memcpy(buffer, &head, sizeof(nty_ophdr));
size = 2;
} else if (length < 0xffff) {
nty_websocket_head_126 hdr = {0};
hdr.payload_length = length;
memcpy(hdr.mask_key, mask, 4);
memcpy(buffer, &head, sizeof(nty_ophdr));
memcpy(buffer+sizeof(nty_ophdr), &hdr, sizeof(nty_websocket_head_126));
size = sizeof(nty_websocket_head_126);
} else {
nty_websocket_head_127 hdr = {0};
hdr.payload_length = length;
memcpy(hdr.mask_key, mask, 4);
memcpy(buffer, &head, sizeof(nty_ophdr));
memcpy(buffer+sizeof(nty_ophdr), &hdr, sizeof(nty_websocket_head_127));
size = sizeof(nty_websocket_head_127);
}
memcpy(buffer+2, stream, length);
return length + 2;
}
//设置非阻塞
static int set_nonblock(int fd) {
int flags;
flags = fcntl(fd, F_GETFL, 0);
if (flags < 0) return -1;
if (fcntl(fd, F_SETFL, flags | O_NONBLOCK) < 0) return -1;
return 0;
}
int send_cb(int fd, int events, void* arg) {
//发送sendbuffer中的数据
struct sockitem* si = (struct sockitem*)arg;
send(fd, si->sendbuffer, si->slen, 0);
//设置关注读事件, 写完这批交互数据，接下来该继续读了
struct epoll_event ev;
ev.events = EPOLLIN | EPOLLET;
si->sockfd = fd;//从新设置sockfd
si->callback = recv_cb;
ev.data.ptr = si;
memset(si->sendbuffer, 0, BUFFSIZE);
//发送完数据从新将缓冲区置为0
epoll_ctl(eventloop->epfd, EPOLL_CTL_MOD, fd, &ev);
}
//关闭连接
int close_connection(struct sockitem* si, unsigned int event) {
struct epoll_event ev;
close(si->sockfd);//关闭连接
//将关注IO事件结点从监视红黑树中删除
ev.events = event;
epoll_ctl(eventloop->epfd, EPOLL_CTL_DEL, si->sockfd, &ev);
free(si);
return 0;
}
//处理读取数据
int recv_cb(int fd, int events, void* arg) {
struct sockitem* si = (struct sockitem*)arg;
struct epoll_event ev;
int ret = recv(fd, si->recvbuffer, BUFFSIZE, 0);
if (ret < 0) {
if (errno == EAGAIN || errno == EWOULDBLOCK) {
return -1;//非阻塞表示缓冲区中没有数据
} else {
}
close_connection(si, EPOLLIN);
} else if (ret == 0) {
printf("disconnect %d\n", fd);
close_connection(si, EPOLLIN);
} else {
si->rlen = 0;//重置rlen
if (si->status == WS_HANDSHARK) {
//说明是请求握手数据
printf("request\n");
printf("%s\n", si->recvbuffer);
handshark(si, eventloop);//完成握手
} else if (si->status == WS_DATATRANSFORM) {
transform(si, eventloop);
} else if (si->status == WS_DATAEND) {
close_connection(si, EPOLLOUT | EPOLLET);
}
}
}
int accept_cb(int fd, int events, void* arg) {
//处理新的连接。 连接IO事件处理流程
struct sockaddr_in cli_addr;
memset(&cli_addr, 0, sizeof(cli_addr));
socklen_t cli_len = sizeof(cli_addr);
int cli_fd = accept(fd, (SA*)&cli_addr, &cli_len);
if (cli_fd <= 0) return -1;
char cli_ip[INET_ADDRSTRLEN] = {0}; //存储cli_ip
printf("recv from ip %s at port %d\n", inet_ntop(AF_INET, &cli_addr.sin_addr, cli_ip, sizeof(cli_ip)),
ntohs(cli_addr.sin_port));
//注册接下来的读事件处理器
struct epoll_event ev;
ev.events = EPOLLIN | EPOLLET;
struct sockitem* si = (struct sockitem*)malloc(sizeof(struct sockitem));
si->sockfd = cli_fd;
si->status = WS_HANDSHARK;//等待握手的状态
si->callback = recv_cb;//设置事件处理器
ev.data.ptr = si;
epoll_ctl(eventloop->epfd, EPOLL_CTL_ADD, cli_fd, &ev);
return cli_fd;
}
int main(int argc, char* argv[]) {
if (argc != 2) {
fprintf(stderr, "usage %s <port>", argv[0]);
return -1;
}
int port = atoi(argv[1]);
int sockfd = socket(AF_INET, SOCK_STREAM, 0);
if (sockfd == -1) {
fprintf(stderr, "socket error");
return -2;
}
set_nonblock(sockfd);
struct sockaddr_in serv_addr;
memset(&serv_addr, 0, sizeof(serv_addr));
serv_addr.sin_family = AF_INET;
serv_addr.sin_addr.s_addr = INADDR_ANY;
serv_addr.sin_port = htons(port);
if (bind(sockfd, (SA*)&serv_addr, sizeof(serv_addr)) == -1) {
fprintf(stderr, "bind error");
return -3;
}
if (listen(sockfd, 5) == -1) {
fprintf(stderr, "listen error");
return -4;
}
//init eventloop
eventloop = (struct reactor*)malloc(sizeof(struct reactor));
//创建监视事件红黑树的根部
eventloop->epfd = epoll_create(1);
//注册处理连接IO处理函数
struct epoll_event ev;
ev.events = EPOLLIN;
struct sockitem* si = (struct sockitem*)malloc(sizeof(struct sockitem));
si->sockfd = sockfd;
si->callback = accept_cb;
ev.data.ptr = si;
epoll_ctl(eventloop->epfd, EPOLL_CTL_ADD, sockfd, &ev);
while (1) {
int nready = epoll_wait(eventloop->epfd, eventloop->events, 512, -1);
if (nready < -1) {
break;
}
int i = 0;
for (; i < nready; ++i) {
if (eventloop->events[i].events & EPOLLIN) {
struct sockitem *si = (struct sockitem*)eventloop->events[i].data.ptr;
si->callback(si->sockfd, eventloop->events[i].events, si);
}
if (eventloop->events[i].events & EPOLLOUT) {
struct sockitem *si = (struct sockitem*)eventloop->events[i].data.ptr;
si->callback(si->sockfd, eventloop->events[i].events, si);
}
}
}
return 0;
} |
线上测试工具 + 我的测试结果
总结本文
本文从http的弊端入手分析为啥需要websocket这个全新的应用层协议出来
为了解决服务器需要向客户端主动推送数据的问题. 后端服务器向前端网页主动推送数据.
http 轮询 长连接 虽然也可以，但是轮询延迟长，而且不断地建立无效连接，结果服务器压根不需要推送数据，这样就很浪费资源, 长轮询，虽说是解决了延迟问题，可是不断地占据着服务器，对于服务器资源也是一种浪费, 毕竟你霸占服务器然而很长事件才需要推送一次数据
于是全新地服务器可以主动推送数据地, 基于tcp地全双工地websocket 诞生了
websocket分为 握手和数据交互两大阶段。 握手阶段是基于http升级的.
为了区分recv的时候的数据阶段，于是状态机诞生了
握手阶段的核心在于，密匙确认服务端是否支持websocket. key----> accept
经过 str += GUID SHA-1(str) hash 然后base64_encode (str);
然后我们基于tcp如果需要封装自己的应用层协议:
特有数据帧格式：1. 操作码 2. 包长度 3, mask-key
|
