好了，现在我们来将服务器的连接做一个简单的封装，在这之前，我们要将我我们所发的数据做一个规定，采用TLV的格式，来进行封装。目的是解决TCP传输的粘包问题。

5.1 Message消息封装

TCP粘包问题-拆包封包过程
先创建一个message.h头文件

lars_reactor/include/message.h

  1. #pragma once
  3. //解决tcp粘包问题的消息头
  4. struct msg_head
  5. {
  6.     int msgid;
  7.     int msglen;
  8. };
  10. //消息头的二进制长度，固定数
  11. #define MESSAGE_HEAD_LEN 8
  13. //消息头+消息体的最大长度限制
  14. #define MESSAGE_LENGTH_LIMIT (65535 - MESSAGE_HEAD_LEN)
  1.     接下来我们每次在server和 client之间传递数据的时候，都发送这种数据格式的头再加上后面的数据内容即可。

5.2 创建一个tcp_conn连接类

lars_reactor/include/tcp_conn.h

  1. #pragma once
  3. #include "reactor_buf.h"
  4. #include "event_loop.h"
  6. //一个tcp的连接信息
  7. class tcp_conn
  8. {
  9. public:
  10.     //初始化tcp_conn
  11.     tcp_conn(int connfd, event_loop *loop);
  13.     //处理读业务
  14.     void do_read();
  16.     //处理写业务
  17.     void do_write();
  19.     //销毁tcp_conn
  20.     void clean_conn();
  22.     //发送消息的方法
  23.     int send_message(const char *data, int msglen, int msgid);
  25. private:
  26.     //当前链接的fd
  27.     int _connfd;
  28.     //该连接归属的event_poll
  29.     event_loop *_loop;
  30.     //输出buf
  31.     output_buf obuf;     
  32.     //输入buf
  33.     input_buf ibuf;
  34. };

简单说明一下里面的成员和方法：

成员:

_connfd：server刚刚accept成功的套接字

_loop:当前链接所绑定的事件触发句柄.

obuf:链接输出缓冲，向对端写数据

ibuf：链接输入缓冲，从对端读数据

方法：

tcp_client()：构造，主要在里面实现初始化及创建链接链接的connect过程。

do_read():读数据处理业务，主要是EPOLLIN事件触发。

do_write():写数据处理业务，主要是EPOLLOUT事件触发。

clean_conn():清空链接资源。

send_message()：将消息打包成TLV格式发送给对端。

  1.     接下来，实现以下`tcp_conn`类.

lars_reactor/src/tcp_conn.cpp

  1. #include <unistd.h>
  2. #include <fcntl.h>
  3. #include <sys/types.h>
  4. #include <sys/socket.h>
  5. #include <netinet/in.h>
  6. #include <netinet/tcp.h>
  7. #include <string.h>
  9. #include "tcp_conn.h"
  10. #include "message.h"
  13. //回显业务
  14. void callback_busi(const char *data, uint32_t len, int msgid, void *args, tcp_conn *conn)
  15. {
  16.     conn->send_message(data, len, msgid);
  17. }
  20. //连接的读事件回调
  21. static void conn_rd_callback(event_loop *loop, int fd, void *args)
  22. {
  23.     tcp_conn *conn = (tcp_conn*)args;
  24.     conn->do_read();
  25. }
  26. //连接的写事件回调
  27. static void conn_wt_callback(event_loop *loop, int fd, void *args)
  28. {
  29.     tcp_conn *conn = (tcp_conn*)args;
  30.     conn->do_write();
  31. }
  33. //初始化tcp_conn
  34. tcp_conn::tcp_conn(int connfd, event_loop *loop)
  35. {
  36.     _connfd = connfd;
  37.     _loop = loop;
  38.     //1. 将connfd设置成非阻塞状态
  39.     int flag = fcntl(_connfd, F_GETFL, 0);
  40.     fcntl(_connfd, F_SETFL, O_NONBLOCK|flag);
  42.     //2. 设置TCP_NODELAY禁止做读写缓存，降低小包延迟
  43.     int op = 1;
  44.     setsockopt(_connfd, IPPROTO_TCP, TCP_NODELAY, &op, sizeof(op));//need netinet/in.h netinet/tcp.h
  46.     //3. 将该链接的读事件让event_loop监控 
  47.     _loop->add_io_event(_connfd, conn_rd_callback, EPOLLIN, this);
  49.     //4 将该链接集成到对应的tcp_server中
  50.     //TODO
  51. }
  53. //处理读业务
  54. void tcp_conn::do_read()
  55. {
  56.     //1. 从套接字读取数据
  57.     int ret = ibuf.read_data(_connfd);
  58.     if (ret == -1) {
  59.         fprintf(stderr, "read data from socket\n");
  60.         this->clean_conn();
  61.         return ;
  62.     }
  63.     else if ( ret == 0) {
  64.         //对端正常关闭
  65.         printf("connection closed by peer\n");
  66.         clean_conn();
  67.         return ;
  68.     }
  70.     //2. 解析msg_head数据    
  71.     msg_head head;    
  73.     //[这里用while，可能一次性读取多个完整包过来]
  74.     while (ibuf.length() >= MESSAGE_HEAD_LEN)  {
  75.         //2.1 读取msg_head头部，固定长度MESSAGE_HEAD_LEN    
  76.         memcpy(&head, ibuf.data(), MESSAGE_HEAD_LEN);
  77.         if(head.msglen > MESSAGE_LENGTH_LIMIT || head.msglen < 0) {
  78.             fprintf(stderr, "data format error, need close, msglen = %d\n", head.msglen);
  79.             this->clean_conn();
  80.             break;
  81.         }
  82.         if (ibuf.length() < MESSAGE_HEAD_LEN + head.msglen) {
  83.             //缓存buf中剩余的数据，小于实际上应该接受的数据
  84.             //说明是一个不完整的包，应该抛弃
  85.             break;
  86.         }
  88.         //2.2 再根据头长度读取数据体，然后针对数据体处理 业务
  89.         //TODO 添加包路由模式
  91.         //头部处理完了，往后偏移MESSAGE_HEAD_LEN长度
  92.         ibuf.pop(MESSAGE_HEAD_LEN);
  94.         //处理ibuf.data()业务数据
  95.         printf("read data: %s\n", ibuf.data());
  97.         //回显业务
  98.         callback_busi(ibuf.data(), head.msglen, head.msgid, NULL, this);
  100.         //消息体处理完了,往后便宜msglen长度
  101.         ibuf.pop(head.msglen);
  102.     }
  104.     ibuf.adjust();
  106.     return ;
  107. }
  109. //处理写业务
  110. void tcp_conn::do_write()
  111. {
  112.     //do_write是触发玩event事件要处理的事情，
  113.     //应该是直接将out_buf力度数据io写会对方客户端 
  114.     //而不是在这里组装一个message再发
  115.     //组装message的过程应该是主动调用
  117.     //只要obuf中有数据就写
  118.     while (obuf.length()) {
  119.         int ret = obuf.write2fd(_connfd);
  120.         if (ret == -1) {
  121.             fprintf(stderr, "write2fd error, close conn!\n");
  122.             this->clean_conn();
  123.             return ;
  124.         }
  125.         if (ret == 0) {
  126.             //不是错误，仅返回0表示不可继续写
  127.             break;
  128.         }
  129.     }
  131.     if (obuf.length() == 0) {
  132.         //数据已经全部写完，将_connfd的写事件取消掉
  133.         _loop->del_io_event(_connfd, EPOLLOUT);
  134.     }
  136.     return ;
  137. }
  139. //发送消息的方法
  140. int tcp_conn::send_message(const char *data, int msglen, int msgid)
  141. {
  142.     printf("server send_message: %s:%d, msgid = %d\n", data, msglen, msgid);
  143.     bool active_epollout = false; 
  144.     if(obuf.length() == 0) {
  145.         //如果现在已经数据都发送完了，那么是一定要激活写事件的
  146.         //如果有数据，说明数据还没有完全写完到对端，那么没必要再激活等写完再激活
  147.         active_epollout = true;
  148.     }
  150.     //1 先封装message消息头
  151.     msg_head head;
  152.     head.msgid = msgid;
  153.     head.msglen = msglen;
  155.     //1.1 写消息头
  156.     int ret = obuf.send_data((const char *)&head, MESSAGE_HEAD_LEN);
  157.     if (ret != 0) {
  158.         fprintf(stderr, "send head error\n");
  159.         return -1;
  160.     }
  162.     //1.2 写消息体
  163.     ret = obuf.send_data(data, msglen);
  164.     if (ret != 0) {
  165.         //如果写消息体失败，那就回滚将消息头的发送也取消
  166.         obuf.pop(MESSAGE_HEAD_LEN);
  167.         return -1;
  168.     }
  170.     if (active_epollout == true) {
  171.         //2. 激活EPOLLOUT写事件
  172.         _loop->add_io_event(_connfd, conn_wt_callback, EPOLLOUT, this);
  173.     }
  176.     return 0;
  177. }
  179. //销毁tcp_conn
  180. void tcp_conn::clean_conn()
  181. {
  182.     //链接清理工作
  183.     //1 将该链接从tcp_server摘除掉    
  184.     //TODO 
  185.     //2 将该链接从event_loop中摘除
  186.     _loop->del_io_event(_connfd);
  187.     //3 buf清空
  188.     ibuf.clear(); 
  189.     obuf.clear();
  190.     //4 关闭原始套接字
  191.     int fd = _connfd;
  192.     _connfd = -1;
  193.     close(fd);
  194. }
  1.     具体每个方法的实现，都很清晰。其中`conn_rd_callback()`和`conn_wt_callback()`是注册读写事件的回调函数，设置为static是因为函数类型没有this指针。在里面分别再调用`do_read()`和`do_write()`方法。

5.3 修正tcp_server对accept之后的处理方法

lars_reactor/src/tcp_server.cpp

  1. //...
  3. //开始提供创建链接服务
  4. void tcp_server::do_accept()
  5. {
  6.     int connfd;    
  7.     while(true) {
  8.         //accept与客户端创建链接
  9.         printf("begin accept\n");
  10.         connfd = accept(_sockfd, (struct sockaddr*)&_connaddr, &_addrlen);
  11.         if (connfd == -1) {
  12.             if (errno == EINTR) {
  13.                 fprintf(stderr, "accept errno=EINTR\n");
  14.                 continue;
  15.             }
  16.             else if (errno == EMFILE) {
  17.                 //建立链接过多，资源不够
  18.                 fprintf(stderr, "accept errno=EMFILE\n");
  19.             }
  20.             else if (errno == EAGAIN) {
  21.                 fprintf(stderr, "accept errno=EAGAIN\n");
  22.                 break;
  23.             }
  24.             else {
  25.                 fprintf(stderr, "accept error");
  26.                 exit(1);
  27.             }
  28.         }
  29.         else {
  30.             //accept succ!
  31.             // ============= 将之前的触发回调的删掉，改成如下====
  32.             tcp_conn *conn = new tcp_conn(connfd, _loop);
  33.             if (conn == NULL) {
  34.                 fprintf(stderr, "new tcp_conn error\n");
  35.                 exit(1);
  36.             }
  37.             // ============================================
  38.             printf("get new connection succ!\n");
  39.             break;
  40.         }
  41.     }
  42. }
  44. //...
  1.     这样，每次accept成功之后，创建一个与当前客户端套接字绑定的tcp_conn对象。在构造里就完成了基本的对于EPOLLIN事件的监听和回调动作.
  3.     现在可以先编译一下，保证没有语法错误，但是如果想测试，就不能够使用`nc`指令测试了，因为现在服务端只能够接收我们自定义的TLV格式的报文。那么我们需要自己写一个客户端来完成基本的测试。