Chinaunix首页 | 论坛 | 博客

justjust.blog.chinaunix.net

首页　| 　博文目录　| 　关于我

justscu

  • 博客访问： 157528
  • 博文数量： 22
  • 博客积分： 425
  • 博客等级： 下士
  • 技术积分： 350
  • 用 户 组： 普通用户
  • 注册时间： 2012-05-15 09:43
个人简介

专注服务器设计、开发； https://github.com/justscu；

文章分类

全部博文（22）

文章存档

2014年（10）

2013年（2）

2012年（10）

我的朋友
最近访客
推荐博文
相关博文
用epoll实现一个异步处理的模型(2)

分类： LINUX

2013-11-07 20:30:42

1. 基于上一篇文章（用epoll实现一个异步处理的模型(1)），来继续实现一个tcp服务器。
    （1）构造一个监听socket(socket-> bind-> listen)，将监听socket的fd加入到epoll的监听中，等待客户端的连接。
    （2）客户端连接后，调用OnFDRead()，得到客户端socket的fd，并将fd加入到epoll的监听中。
    （3）处理客户端fd的读写事件。
2. 代码

点击(此处)折叠或打开  tcpeventhandler.h

  1. #pragma once
  2. #include "reactor.h"
  3. #include <sys/types.h> /* See NOTES */
  4. #include <sys/socket.h>
  5. #include <errno.h>
  6. #include <arpa/inet.h>
  7. #include <netinet/tcp.h>
  8. #include <fcntl.h>
  9. #include <string.h>
  10. #include <stdio.h>

  12. class SocketDataDecoderBase
  13. {
  14. public:
  15.     virtual int OnProcess(FDEventHanderBase* pSocket, const char* buf,
  16.      unsigned int buf_len) = 0;
  17. };

  19. /********************************
  20.  * 处理tcpServer事件的类
  21.  ********************************/
  22. class TcpServerEventHandler: public FDEventHanderBase
  23. {
  24. public:
  25.     TcpServerEventHandler(const char* ip, int port, SocketDataDecoderBase* decoder,
  26.      int fd, Reactor* pReactor) :
  27.         FDEventHanderBase(fd, pReactor), m_pDecoder(decoder)
  28.     {
  29.         memset(m_ip, 0x00, sizeof(m_ip));
  30.         strncpy(m_ip, ip, strlen(ip));
  31.         m_port = port;
  32.     }
  33.     virtual ~TcpServerEventHandler() {
  34.     }
  35.     // create listen socket. return 0, success.
  36.     int OnListen();
  37.     int OnAccept(int fd);
  38.     // 当epoll中有新的连接过来的时候，就会调用该函数
  39.     virtual void OnFDRead();
  40.     virtual void OnFDWrite();
  41.     virtual void Close();
  42. protected:
  43.     char m_ip[16]; //ipv4
  44.     int m_port;
  45.     SocketDataDecoderBase* m_pDecoder;
  46. };

  48. /* **************************
  49.  * 处理TCP客户端发过来的数据
  50.  * 这个地方，可以开辟2块缓冲区（读缓冲区和写缓冲区）
  51.  * (1)读，当socket可读时，epoll会被触发，回调OnFDRead()，
  52.  * 将读取到的数据进行处理，或者放到读缓冲区
  53.  * (2)写，当有数据需要通过socket发出去时，先将数据放到写缓冲区中，
  54.  * 注册RegisterWriteEvent，之后,OnFDWrite就会被回调，
  55.  * 若LT模式，当数据没有写完时，OnFDWrite一直会被回调，直到写完后，
  56.  * 用户调用UnRegisteWriteEvent.
  57.  * ************************/
  58. class TcpSocketHandler: public FDEventHanderBase
  59. {
  60. public:
  61.     TcpSocketHandler(SocketDataDecoderBase* decoder, int fd, Reactor* pReactor) :
  62.         FDEventHanderBase(fd, pReactor), m_dataDecoder(decoder)
  63.     {

  65.     }
  66.     virtual ~TcpSocketHandler() {
  67.     }
  68.     // 读被触发时，调用该函数，处理客户端发过来的数据
  69.     virtual void OnFDRead();
  70.     // 写被触发时，调用该函数
  71.     virtual void OnFDWrite();
  72.     // 关闭fd
  73.     virtual void Close();
  74.     // 可以通过该函数，向客户端发送数据
  75.     virtual int SendBuf(const char* buf, int buf_len);
  76. private:
  77.     SocketDataDecoderBase* m_dataDecoder;
  78.     //Buffer* pReadBuf;
  79.     //Buffer* pWriteBuf;
  80. };

  82. /* *********************************
  83.  *          TcpServerBase类
  84.  * *********************************/
  85. class TcpServerBase
  86. {
  87. public:
  88.     TcpServerBase(const char* ip, int port, SocketDataDecoderBase* pDecoder)
  89.     {
  90.         memset(m_ip, 0x00, sizeof(m_ip));
  91.         strncpy(m_ip, ip, strlen(ip));
  92.         m_port = port;
  93.         m_pReactor = NULL;
  94.         m_pDecoder = pDecoder;
  95.     }
  96.     virtual int Init(){
  97.         m_pReactor = new LTReactor;
  98.         return m_pReactor->Init();
  99.     }
  100.     virtual int Run(){
  101.         TcpServerEventHandler h(m_ip, m_port, m_pDecoder, 0, m_pReactor);
  102.         if (0 == h.OnListen()) {
  103.             m_pReactor->Run();
  104.         }

  106.         return 0;
  107.     }
  108.     virtual int Exit(){
  109.         delete m_pReactor;
  110.         return 0;
  111.     }
  112. private:
  113.     char m_ip[16];//ipv4
  114.     int m_port;
  115.     Reactor* m_pReactor;
  116.     SocketDataDecoderBase* m_pDecoder;
  117. };

点击(此处)折叠或打开 tcpeventhandler.cpp

  1. #include "tcpeventhandler.h"

  3. /********************************
  4.  * 处理tcpServer事件的类
  5.  ********************************/
  6. // create listen socket. return 0, success.
  7. int TcpServerEventHandler::OnListen() {
  8.     m_fd = ::socket(AF_INET, SOCK_STREAM, 0);
  9.     if (m_fd == -1)
  10.     {
  11.         std::cout << "create socket error." << strerror(errno) << std::endl;
  12.         return -1;
  13.     }
  14.     sockaddr_in addr;
  15.     memset(&addr, 0x00, sizeof(sockaddr_in));
  16.     addr.sin_family = AF_INET;
  17.     addr.sin_port = htons(m_port);
  18.     socklen_t len = sizeof(sockaddr_in);

  20.     if (1 != inet_pton(AF_INET, m_ip, &addr.sin_addr))
  21.     {
  22.         std::cout << "inet_pton error." << strerror(errno) << std::endl;
  23.         close(m_fd);
  24.         return -1;
  25.     }
  26.     // 允许在同一端口上启动同一服务器的多个实例
  27.     int opVal = 1;
  28.     if (0 != setsockopt(m_fd, SOL_SOCKET, SO_REUSEADDR, &opVal, sizeof(opVal)))
  29.     {
  30.         std::cout << "socketopt error." << strerror(errno) << std::endl;
  31.         close(m_fd);
  32.         return -1;
  33.     }
  34.     // 设置非阻塞模式
  35.     SetNonBlock(m_fd, true);
  36.     // bind
  37.     if (-1 == bind(m_fd, (struct sockaddr*) &addr, len))
  38.     {
  39.         std::cout << "bind error." << strerror(errno) << std::endl;
  40.         close(m_fd);
  41.         return -1;
  42.     }
  43.     // listen
  44.     if (-1 == listen(m_fd, 200))
  45.     {
  46.         std::cout << "listen error." << strerror(errno) << std::endl;
  47.         close(m_fd);
  48.         return -1;
  49.     }

  51.     // 将该fd加入到 epoll中，监听客户端的连接
  52.     if (0 != RegisterReadEvent())
  53.     {
  54.         std::cout << "RegisterReadEvent error." << std::endl;
  55.         close(m_fd);
  56.         return -1;
  57.     }

  59.     return 0;
  60. }

  62. int TcpServerEventHandler::OnAccept(int fd) {
  63.     TcpSocketHandler* pHandler = new TcpSocketHandler(m_pDecoder, fd,
  64.      this->m_pReactor);
  65.     // 注册一个读事件，从TCP的客户端读取数据.
  66.     if (0 != pHandler->RegisterReadEvent())
  67.     {
  68.         pHandler->Close();
  69.         delete pHandler;
  70.         pHandler = NULL;
  71.     }
  72.     // 需要delete pHandler
  73.     return 0;
  74. }
  75. // 当epoll中有新的连接过来的时候，就会调用该函数
  76. void TcpServerEventHandler::OnFDRead() {
  77.     sockaddr_in clientAddr;
  78.     socklen_t len = sizeof(clientAddr);
  79.     // 客户端的fd，这个地方，其实可以建立一个(fd -- 客户端)对应关系
  80.     int sockFD = accept(m_fd, (struct sockaddr*) &clientAddr, &len);
  81.     if (sockFD < 0)
  82.     {
  83.         std::cout << "accept error." << std::endl;
  84.         return;
  85.     }
  86.     else
  87.     {
  88.         int port = ntohs(clientAddr.sin_port);
  89.         char addr[256] =
  90.         { 0 };
  91.         inet_ntop(AF_INET, &clientAddr.sin_addr.s_addr, addr,
  92.          (socklen_t) sizeof(addr));
  93.         printf("accept: [%s:%d] \n", addr, port);
  94.     }

  96.     OnAccept(sockFD);
  97. }
  98. void TcpServerEventHandler::OnFDWrite() {

  100. }
  101. void TcpServerEventHandler::Close() {
  102.     UnRegisterReadEvent();
  103.     UnRegisterWriteEvent();
  104.     close(m_fd);
  105. }

  107. /* ***********************
  108.  * 处理TCP客户端发过来的数据
  109.  * ************************/
  110. // 读被触发时，调用该函数，处理客户端发过来的数据
  111. void TcpSocketHandler::OnFDRead() {
  112.     // 在这个函数中，可以读取客户端发过来的数据，并对数据进行分析，TCP的数据，
  113.     // 可能会出现半包的情况，所以一般会有一个缓存来保存没有处理完毕的数据
  114.     char buf[1024] =
  115.     { 0 };
  116.     int iRet = read(m_fd, buf, sizeof(buf));
  117.     if (0 == iRet)
  118.     {
  119.         Close();// 客户端关闭socket.
  120.         return;
  121.     }
  122.     // 回调处理函数，对数据进行处理
  123.     m_dataDecoder->OnProcess(this, buf, iRet);
  124. }
  125. // 写被触发时，调用该函数
  126. void TcpSocketHandler::OnFDWrite() {

  128. }
  129. // 关闭fd
  130. void TcpSocketHandler::Close() {
  131.     UnRegisterReadEvent();
  132.     UnRegisterWriteEvent();
  133.     close(m_fd);
  134.     delete this;
  135. }
  136. // 可以通过该函数，向客户端发送数据
  137. int TcpSocketHandler::SendBuf(const char* buf, int buf_len) {
  138.     RegisterWriteEvent();
  139.     int iRet = write(m_fd, buf, buf_len);
  140.     UnRegisterWriteEvent();
  141.     return iRet;
  142. }
3.  测试代码
    测试代码实现了一个基本的tcp回显服务器。
    简单的测试，可以在命令行，nc localhost 56893，然后输入，就能够看到回显的效果。

测试代码(此处)

  1. #include "tcpeventhandler.h"

  3. /* *********************
  4.  * 处理客户端发过来的数据
  5.  * ********************/
  6. class TcpDataDecoder_echo_test: public SocketDataDecoderBase
  7. {
  8. public:
  9.     // 真正的处理数据的函数
  10.     virtual int OnProcess(FDEventHanderBase* pSocket, const char* buf,
  11.      unsigned int buf_len)
  12.     {
  13.         // 这个地方，将客户端的数据回写
  14.         if (0 == pSocket->RegisterWriteEvent())
  15.         {
  16.             write(pSocket->GetFD(), buf, buf_len);
  17.             pSocket->UnRegisterWriteEvent();
  18.         }

  20.         return 0;
  21.     }
  22. };


  25. void TcpServer_test() {
  26.     SocketDataDecoderBase* pDecoder = new TcpDataDecoder_echo_test;
  27.     TcpServerBase b("127.0.0.1", 56893, pDecoder);
  28.     b.Init();
  29.     b.Run();
  30.     b.Exit();
  31.     delete pDecoder;

  33. }
4. 一些需要注意的地方
    （1）服务器端，最好设置tcp keepalive，在服务器与客户端之间，维持一个心跳连接。防止一些假死的客户端占用服务器端资源。【上面的代码中，并没有体现出来】
    （2）由于服务器端的fd，是连续回收使用的，所以当close(fd) 后，需要将该fd的已经注册的读写handler，从EventHanderSet中删除掉。
    （3）在epoll的LT模式下，在写之前，需要注册RegistReadEvent()，之后，该fd可以一直写。当写完后，需要调用UnRegisterReadEvent()。
    （4）通常的接收和发送，是需要接收/发送缓存的，上面的代码中，并没有体现出来。
5. 同样的，可以使用unix_socket连接。
    unix socket，用于同一主机上不同进程间的交互。通过socket文件进行交互。

点击(此处)折叠或打开 unixsocketeventhandler.h

  1. #pragma once

  3. #include <stdlib.h>
  4. #include <stdio.h>
  5. #include <string.h>
  6. #include <sys/types.h> /* See NOTES */
  7. #include <sys/socket.h>
  8. #include <netinet/in.h>
  9. #include <sys/un.h>
  10. #include <iostream>
  11. #include <errno.h>
  12. #include <sys/stat.h>

  14. #include "reactor.h"
  15. #include "SocketDataDecoderBase.h"

  17. #define MAX_PATH 256

  19. /* ********************************
  20.  *    处理 unix socket server事件
  21.  * ********************************/
  22. class UnixSocketServerEventHandler: public FDEventHanderBase
  23. {
  24. public:
  25.     UnixSocketServerEventHandler(const char* path, SocketDataDecoderBase* decoder, int fd, Reactor* pReactor) :
  26.         FDEventHanderBase(fd, pReactor), m_pDecoder(decoder)
  27.     {
  28.         assert(strlen(path) > 0);
  29.         strncpy(m_path, path, sizeof(m_path));
  30.         unlink(m_path);
  31.     }
  32.     ~UnixSocketServerEventHandler() {

  34.     }
  35.     // create unix socket. return 0, success.
  36.     int OnListen();
  37.     int OnAccept(int fd);
  38.     // 读被触发时，调用该函数
  39.     virtual void OnFDRead();
  40.     // 写被触发时，调用该函数
  41.     virtual void OnFDWrite() {

  43.     }
  44.     // 关闭fd
  45.     virtual void Close() {
  46.         unlink(m_path);
  47.         close(m_fd);
  48.     }
  49. private:
  50.     char m_path[MAX_PATH];
  51.     SocketDataDecoderBase* m_pDecoder;
  52. };

  54. /* ********************************
  55.  *    处理 unix socket 客户端事件
  56.  * ********************************/
  57. class UnixSocketHandler: public FDEventHanderBase
  58. {
  59. public:
  60.     UnixSocketHandler(SocketDataDecoderBase* decoder, int fd, Reactor* pReactor) :
  61.         FDEventHanderBase(fd, pReactor), m_pDecoder(decoder)
  62.     {

  64.     }
  65.     ~UnixSocketHandler(){}
  66.     virtual void OnFDRead();
  67.     virtual void OnFDWrite();
  68.     virtual void Close();
  69.     virtual int SendBuf(const char* buf, int buf_len);
  70. private:
  71.     SocketDataDecoderBase* m_pDecoder;
  72. };

  74. /* *********************************
  75.  *          UnixSocketServerBase类
  76.  * *********************************/
  77. class UnixSocketServerBase
  78. {
  79. public:
  80.     explicit UnixSocketServerBase(const char* path, SocketDataDecoderBase* pDecoder)
  81.     {
  82.         memset(m_path, 0x00, sizeof(m_path));
  83.         strncpy(m_path, path, strlen(path));
  84.         m_pReactor = NULL;
  85.         m_pDecoder = pDecoder;
  86.     }
  87.     virtual int Init()
  88.     {
  89.         m_pReactor = new LTReactor;
  90.         return m_pReactor->Init();
  91.     }
  92.     virtual int Run()
  93.     {
  94.         UnixSocketServerEventHandler h(m_path, m_pDecoder, 0, m_pReactor);
  95.         if (0 == h.OnListen())
  96.         {
  97.             m_pReactor->Run();
  98.         }

  100.         return 0;
  101.     }
  102.     virtual int Exit()
  103.     {
  104.         delete m_pReactor;
  105.         return 0;
  106.     }
  107. private:
  108.     UnixSocketServerBase(const UnixSocketServerBase&);
  109.     const UnixSocketServerBase& operator=(const UnixSocketServerBase&);
  110. private:
  111.     char m_path[MAX_PATH];
  112.     Reactor* m_pReactor;
  113.     SocketDataDecoderBase* m_pDecoder;
  114. };

点击(此处)折叠或打开 unixsocketeventhandler.cpp

  1. #include "unixsocketeventhandler.h"

  3. // create unix socket. return 0, success.
  4. int UnixSocketServerEventHandler::OnListen() {
  5.     m_fd = socket(AF_UNIX, SOCK_STREAM, 0);
  6.     if (m_fd == -1)
  7.     {
  8.         std::cout << "create socket error." << strerror(errno) << std::endl;
  9.         return -1;
  10.     }

  12.     sockaddr_un addr;
  13.     memset(&addr, 0, sizeof(addr));
  14.     addr.sun_family = AF_UNIX;
  15.     strcpy(addr.sun_path, m_path);

  17.     // 设置非阻塞模式
  18.     SetNonBlock(m_fd, true);
  19.     // 在bind的时候，会生成socket文件。
  20.     socklen_t len = offsetof(struct sockaddr_un, sun_path)+strlen(addr.sun_path);
  21.     if (bind(m_fd, (struct sockaddr *) &addr, len) != 0)
  22.     {
  23.         std::cout << "unix socket bind error,path = " << m_path << std::endl;
  24.         return -1;
  25.     }

  27.     if (listen(m_fd, 500) != 0)
  28.     {
  29.         std::cout << "unix listen error" << std::endl;
  30.         return -1;
  31.     }
  32.     // 更改socket文件的访问权限.
  33.     if (0 != chmod(m_path, 0777))
  34.     {
  35.         std::cout << "chmod[777] " << m_path << "error " << strerror(errno)
  36.          << std::endl;
  37.         return -1;
  38.     }

  40.     // 将该fd加入到 epoll中，监听客户端的连接
  41.     if (0 != RegisterReadEvent())
  42.     {
  43.         std::cout << "RegisterReadEvent error." << std::endl;
  44.         close(m_fd);
  45.         return -1;
  46.     }
  47.     return 0;
  48. }

  50. // 读被触发时，调用该函数
  51. void UnixSocketServerEventHandler::OnFDRead() {
  52.     sockaddr_un clientAddr;
  53.     memset(&clientAddr, 0x00, sizeof(sockaddr_un));
  54.     socklen_t len = sizeof(clientAddr);
  55.     // 客户端的fd，这个地方，其实可以建立一个(fd -- 客户端)对应关系
  56.     int sockFD = accept(m_fd, (struct sockaddr*) &clientAddr, &len);
  57.     if (sockFD < 0)
  58.     {
  59.         std::cout << "accept error." << std::endl;
  60.         return;
  61.     }
  62.     else
  63.     {
  64.         len -= offsetof(struct sockaddr_un, sun_path);
  65.         clientAddr.sun_path[len] = 0;
  66.         std::cout << "accpet path = " << clientAddr.sun_path << std::endl;
  67.     }

  69.     OnAccept(sockFD);
  70. }

  72. int UnixSocketServerEventHandler::OnAccept(int fd) {
  73.     SetNonBlock(fd, true);
  74.     UnixSocketHandler* pHandler = new UnixSocketHandler(m_pDecoder, fd,
  75.      this->m_pReactor);
  76.     if (pHandler == NULL)
  77.     {
  78.         return -1;
  79.     }
  80.     if (0 != pHandler->RegisterReadEvent())
  81.     {
  82.         pHandler->Close();
  83.         delete pHandler;
  84.         pHandler = NULL;
  85.         return -1;
  86.     }
  87.     return 0;
  88. }

  90. /* ********************************
  91.  *    处理 unix socket 客户端事件
  92.  * ********************************/
  93. void UnixSocketHandler::OnFDRead() {
  94.     char buf[1024] =
  95.     { 0 };
  96.     int iRet = read(m_fd, buf, sizeof(buf));
  97.     if (0 == iRet)
  98.     {
  99.         Close();// 客户端关闭socket.
  100.         return;
  101.     }
  102.     // 回调处理函数，对数据进行处理
  103.     m_pDecoder->OnProcess(this, buf, iRet);
  104. }

  106. void UnixSocketHandler::Close() {
  107.     UnRegisterReadEvent();
  108.     UnRegisterWriteEvent();
  109.     close(m_fd);
  110.     delete this;
  111. }

  113. void UnixSocketHandler::OnFDWrite() {

  115. }

  117. // 可以通过该函数，向客户端发送数据
  118. int UnixSocketHandler::SendBuf(const char* buf, int buf_len) {
  119.     RegisterWriteEvent();
  120.     int iRet = write(m_fd, buf, buf_len);
  121.     UnRegisterWriteEvent();
  122.     return iRet;
  123. }

点击(此处)折叠或打开 测试代码： unixsocketeventhandler_test.cpp

  1. #include "unixsocketeventhandler.h"


  4. class UnixSocketDataDecoder_echo_test: public SocketDataDecoderBase
  5. {
  6. public:
  7.     virtual int OnProcess(FDEventHanderBase* pSocket, const char* buf,
  8.          unsigned int buf_len)
  9.     {
  10.         std::cout << "RecvString: " << buf << std::endl;
  11.         return ((UnixSocketHandler*)pSocket)->SendBuf(buf, buf_len);
  12.     }
  13. };

  15. void UnixSocketEventHandler_test()
  16. {
  17.     const char* path = "/home/ll/work/min_heap/unix_path.sock";
  18.     UnixSocketDataDecoder_echo_test* pDecoder = new UnixSocketDataDecoder_echo_test;
  19.     UnixSocketServerBase server(path, pDecoder);
  20.     server.Init();
  21.     server.Run();
  22.     server.Exit();

  24.     delete pDecoder;
  25.     pDecoder = NULL;
  26. }

  28. // 客户端测试程序
  29. int f_unix_socket_client()
  30. {
  31.     const char* path = "/home/ll/work/min_heap/unix_path.sock";
  32.     int sockFd = socket(AF_UNIX, SOCK_STREAM, 0);
  33.     if ( sockFd < 0 )
  34.     {
  35.         std::cout << "socket Error:" << strerror(errno) << std::endl;
  36.         return -1;
  37.     }

  39.     sockaddr_un server;
  40.     server.sun_family = AF_UNIX;
  41.     strcpy(server.sun_path, path);
  42.     // 直接去connect就好了，不需要再bind，否则会再生成一个socket文件。
  43.     int iRet = connect(sockFd, (struct sockaddr*) &server, sizeof(server));
  44.     if ( 0 != iRet )
  45.     {
  46.         std::cout << "connect Error:" << strerror(errno) << std::endl;
  47.         close(sockFd);
  48.         return 0;
  49.     }
  50.     while(true)
  51.     {
  52.         char inBuf[1024] = {0};
  53.         std::cin >> inBuf;
  54.         write(sockFd, inBuf, strlen(inBuf));

  56.         char outBuf[1024] = {0};
  57.         read(sockFd, outBuf, sizeof(outBuf));
  58.         std::cout << "Echo: " << outBuf << std::endl;
  59.     }

  61.     return 0;
  62. }





阅读(4207) | 评论(0) | 转发(0) |
0

上一篇：用epoll实现一个异步处理的模型(1)

下一篇：消息队列ActiveMQ

给主人留下些什么吧！~~
关于我们 | 关于IT168 | 联系方式 | 广告合作 | 法律声明 | 免费注册

Copyright 2001-2010 ChinaUnix.net All Rights Reserved 北京皓辰网域网络信息技术有限公司. 版权所有

感谢所有关心和支持过ChinaUnix的朋友们

16024965号-6