一、概述
非阻塞网络编程无疑成了高并发、高性能编程的代名词,但现实应用编程中并不是每种应用都需要采用非阻塞编程模式,因为这将大大增加编程的复杂性、开发周期以及出错率,所以我们写的绝大部分网络程序程序都是阻塞的,一般是一个进程一个网络连接或一个线程一个网络连接。即然非阻塞模式可以实现高并发网络连接,阻塞模式可以实现复杂的业务逻辑,那是否有办法将二者结合起来呢?答案是肯定的,其中在 acl_cpp 库中,ipc 目录下的模块就是为了满足这种需求而设计的。
在文章《acl_cpp 非阻塞模块的IPC通信机制》和《ipc_service 类:阻塞与非阻塞混合编程》中分别以两种方式介绍了 如何通过 acl_cpp IPC 通信库来达到阻塞与非阻塞两种编程模式的结合目的,但这篇文章讲的内容还是有点麻烦,用户用也很难上手。于是本人在那两篇文章所讲的库的基础上进一步抽象,实现了另外一个阻塞与非阻塞结合的库。
二、接口设计
在 rpc.hpp 头文件中有两个类:rpc_service 和 rpc_request,其中 rpc_request 是一个纯虚类,用户需要继承此类并实现该类中规定的纯虚接口,从而实现自己的阻塞操作功能;rpc_service 是阻塞与非阻塞结合的粘合类,通过将 rpc_request 子类实例传递给 rpc_service 类实例的 rpc_fork 方法,实现将阻塞请求过程交给子线程处理的目的。
1、rpc_service 类
1.1、构造函数:
/** * 构造函数 * @param nthread {int} 如果该值 > 1 则内部自动采用线程池,否则 * 则是一个请求一个线程 * @param win32_gui {bool} 是否是窗口类的消息,如果是,则内部的 * 通讯模式自动设置为基于 WIN32 的消息,否则依然采用通用的套接 * 口通讯方式 */ rpc_service(int nthread, bool win32_gui = false)
从参数 win32_gui 可以看出,acl_cpp 的阻塞/非阻塞结合模式不仅可以用在通常的网络编程中,同时还可以用在阻塞过程与 WINDOWS 界面消息相结合的方面,这无非是经常用 MFC 进行界面编程的福音(例如,用户 VC 写了一个界面程序---当然这个界面窗口是基于 WIN32 消息的,但如果想进行一些数据库操作或文件下载操作,对用户而言阻塞式方法是非常容易实现这两类需求的,但 WIN32 的界面过程不能堵在任何一个数据库操作或文件下载过程,原来 VC 程序员通常的做法也是创建单独的线程进行阻塞操作,然后通过给主窗口传递消息将操作结果通知至主线程,幸运的是 acl_cpp 的 rpc 相关类可以使这一过程更为方便快捷;再如,你写的一个网络服务器程序的主线程是非阻塞的,但其中你不得不调用别人提供的库以实现用户身份验证的功能,同时这个用于用户认证的库又恰恰是阻塞的---一般也是如此,固然,你也许可以费很周折实现这一过程,同样,acl_cpp 的 rpc 相关类可以帮你解决这类问题)。
1.2、将阻塞处理过程的对象交由子线程处理:
/** * 主线程中运行:将请求任务放入子线程池的任务请求队列中,当线程池 * 中的一个子线程接收到该任务后便调用 rpc_request::rpc_run 方法调 * 用子类的方法,当任务处理完毕后给主线程发消息,在主线程中再调用 * rpc_request::rpc_callback * @param req {rpc_request*} rpc_request 子类实例,非空 */ void rpc_fork(rpc_request* req);
通过此接口,可以将阻塞请求过程交给子线程处理,子线程处理完后再通知主线程。
2、rpc_request 类
在类 rpc_request 中有三个虚接口,用户子类必须实现其中的两个纯虚接口:rpc_run 和 rpc_onover,同时用户可以根据需要实现另一非纯虚接口:rpc_wakeup。是当用户调完 rpc_service::rpc_fork 且子线程接收到此请求任务后 rpc_request::rpc_run 方法会被调用(一定切记:rpc_fork 是在主线程中被调用的,而 rpc_run 是在子线程中被调用的);当 rpc_request::rpc_run 函数返回后,rpc_reuqest::rpc_onover 会在主线程中被调用以表示子线程已经处理完毕(同样需要严重注意:rpc_request::rpc_onover 方法又回到主线程中被调用),这样,通过这两个过程就实现了将阻塞过程放在子线程中处理,主线程的非阻塞过程(非阻塞网络事件或非阻塞的 WIN32 消息过程)异步地等待子线程完成任务。
/** * 在子线程中被调用,子类必须实现此接口,用于处理具体任务 */ virtual void rpc_run(void) = 0; /** * 在主线程中被调用,子类必须实现此接口, * 当子线程处理完请求任务后该接口将被调用,所以该类对象只能 * 是当本接口调用后才能被释放,禁止在调用本接口前释放本类对象 */ virtual void rpc_onover(void) = 0;
另外,在 rpc_request 类中还有一个非纯虚方法:rpc_wakeup,这是做什么用的呢?可以假设这种应用场景,在子线程中调用 rpc_request::rpc_run 方法内部过程中,用户如果需要通知主线程一些中间状态(比如,文件下载的进度)该怎么办?那就在 rpc_run 方法内先调用 rpc_request::rpc_signal 通知主线程子线程处理的中间状态,则在主线程中用户实现的 rpc_request::rpc_wakeup 虚接口就会被调用。下面是 rpc_signal 和 rpc_wakeup 的接口说明:
/** * 在子线程中被调用,内部自动支持套接口或 WIN32 窗口消息 * 子类实例的 rpc_run 方法中可以多次调用此方法向主线程的 * 本类实例发送消息,主线程中调用本对象 rpc_wakeup 方法 * @param ctx {void*} 传递的参数指针,一般应该是动态地址 * 比较好,这样可以避免同一个参数被重复覆盖的问题 */ void rpc_signal(void* ctx); /** * 虚接口:当子线程调用本对象的 rpc_signal 时,在主线程中会 * 调用本接口,通知在任务未完成前(即调用 rpc_onover 前)收到 * 子线程运行的中间状态信息;内部自动支持套接口或 WIN32 窗口 * 消息;应用场景,例如,对于 HTTP 下载应用,在子线程中可以 * 一边下载,一边向主线程发送(调用 rpc_signal 方法)下载进程, * 则主线程会调用本类实例的此方法来处理此消息 */ virtual void rpc_wakeup(void* ctx) { (void) ctx; }
紧接着这个应用场景,假设在子线程调用 rpc_run 的内部通过 rpc_signal 通知主线程的中间状态后,希望主线程能收到此通知消息并且希望得到主线程下一步希望执行的指令才会进一步继续执行。于是便有了 rpc_request::cond_wait 和 rpc_request::cond_signal 两个方法的产生,即子线程通过 cond_wait 阻塞地等待主线程的下一步操作指令,主线程则调用 cond_signal 通知子线程下步指令,下面是这两个方法的说明:
/** * 当子线程调用 rpc_signal 给主线程后,调用本方法可以等待 * 主线程发来下一步指令 * @param timeout {int} 等待超时时间(毫秒),当该值为 0 时 * 则采用非阻塞等待模式,当该值为 < 0 时,则采用完全阻塞 * 等待模式(即一直等到主线程发送 cond_signal 通知),当该 * 值 > 0 时,则等待的最大超时时间为 timeout 毫秒 * @return {bool} 返回 true 表示收到主线程发来的通知信号, * 否则,需要调用 cond_wait_timeout 判断是否是超时引起的 */ bool cond_wait(int timeout = -1); /** * 在子线程中被调用,内部自动支持套接口或 WIN32 窗口消息 * 子类实例的 rpc_run 方法中可以多次调用此方法向主线程的 * 本类实例发送消息,主线程中调用本对象 rpc_wakeup 方法 * @param ctx {void*} 传递的参数指针,一般应该是动态地址 * 比较好,这样可以避免同一个参数被重复覆盖的问题 */ void rpc_signal(void* ctx);
三、示例
如果您能大体明白上面有关 rpc_service 和 rpc_request 类的功能说明,相信下面的例子您也一定能看明白:
// rpc_download.cpp : 定义控制台应用程序的入口点。 // #include "stdafx.h" #include <assert.h> #include "lib_acl.hpp" using namespace acl; typedef enum { CTX_T_CONTENT_LENGTH, CTX_T_PARTIAL_LENGTH, CTX_T_END } ctx_t; struct DOWN_CTX { ctx_t type; int length; }; static int __download_count = 0; class http_download : public rpc_request { public: http_download(aio_handle& handle, const char* addr, const char* url) : handle_(handle) , addr_(addr) , url_(url) , error_(false) , total_read_(0) , content_length_(0) {} ~http_download() {} protected: // 子线程处理函数 void rpc_run() { http_request req(addr_); // HTTP 请求对象 // 设置 HTTP 请求头信息 req.request_header().set_url(url_.c_str()) .set_content_type("text/html") .set_host(addr_.c_str()) .set_method(HTTP_METHOD_GET); // 测试用,显示 HTTP 请求头信息内容 string header; req.request_header().build_request(header); printf("request: %s\r\n", header.c_str()); // 发送 HTTP 请求数据 if (req.request(NULL, 0) == false) { printf("send request error\r\n"); error_ = false; return; } // 获得 HTTP 请求的连接对象 http_client* conn = req.get_client(); assert(conn); DOWN_CTX* ctx = new DOWN_CTX; ctx->type = CTX_T_CONTENT_LENGTH; // 获得 HTTP 响应数据的数据体长度 ctx->length = (int) conn->body_length(); content_length_ = ctx->length; // 通知主线程 rpc_signal(ctx); char buf[8192]; while (true) { // 读 HTTP 响应数据体 int ret = req.get_body(buf, sizeof(buf)); if (ret <= 0) { ctx = new DOWN_CTX; ctx->type = CTX_T_END; ctx->length = ret; // 通知主线程 rpc_signal(ctx); break; } ctx = new DOWN_CTX; ctx->type = CTX_T_PARTIAL_LENGTH; ctx->length = ret; // 通知主线程 rpc_signal(ctx); } } // 主线程处理过程,收到子线程任务完成的消息 void rpc_onover() { printf("%s: read over now, total read: %d, content-length: %d\r\n", addr_.c_str(), total_read_, content_length_); // 当 HTTP 响应都完成时,通知主线程停止事件循环过程 __download_count--; if (__download_count == 0) handle_.stop(); } // 主线程处理过程,收到子线程的通知消息 void rpc_wakeup(void* ctx) { DOWN_CTX* down_ctx = (DOWN_CTX*) ctx; switch (down_ctx->type) { case CTX_T_CONTENT_LENGTH: printf("%s: content-length: %d\r\n", addr_.c_str(), down_ctx->length); break; case CTX_T_PARTIAL_LENGTH: total_read_ += down_ctx->length; printf("%s: partial-length: %d, total read: %d\r\n", addr_.c_str(), down_ctx->length, total_read_); break; case CTX_T_END: printf("%s: read over\r\n", addr_.c_str()); break; default: printf("%s: ERROR\r\n", addr_.c_str()); break; } delete down_ctx; } private: aio_handle& handle_; string addr_; string url_; bool error_; int total_read_; int content_length_; }; static void run(void) { aio_handle handle; rpc_service* service = new rpc_service(10); // 创建 rpc 服务对象 // 打开消息服务器 if (service->open(&handle) == false) { printf("open service error: %s\r\n", last_serror()); return; } // 下载页面内容 http_download down1(handle, "www.sina.com.cn:80", "http://www.sina.com.cn/"); service->rpc_fork(&down1); // 发起一个阻塞会话过程 __download_count++; http_download down2(handle, "www.hexun.com:80", "/"); service->rpc_fork(&down2); // 发起第二个阻塞会话过程 __download_count++; // 异步事件循环过程 while (true) { if (handle.check() == false) break; } delete service; handle.check(); // 保证释放所有延迟关闭的异步对象 } int main(void) { #ifdef WIN32 acl_cpp_init(); #endif run(); printf("Enter any key to continue\r\n"); getchar(); return 0; }
参考:
3)更多文章:http://zsxxsz.iteye.com/
4)acl 下载:
http://sourceforge.net/projects/acl/
svn:svn checkout svn://svn.code.sf.net/p/acl/code/trunk acl-code
github:https://github.com/acl-dev/acl
5) QQ 群:242722074
6) 国内镜像:http://git.oschina.net/acl-dev/acl/tree/master
7) 微博:http://weibo.com/zsxxsz/
相关推荐
adb_cpp adb_cpp
如果你想将 PHPRPC 用于 Delphi.NET,请参见 PHPRPC for .NET 章节的相关内容。 PHPRPC for Delphi 的安装 PHPRPC for Delphi 客户端 PHPRPC for Delphi 对容器类型的支持 PHPRPC for Delphi 对自定义类型的支持 ...
基于mysql及acl开源接口的封装操作,提供相关接口操作实例。
学习c++类的多态、析构、继承等相关代码的cpp文件
async_web_server_cpp在ros1环境下的源代码包,下载可以直接编译使用,详细用法可以参看相关文章
图书馆管理的几个类 BookITem ReaderBase 及其相关子类
async_web_server_cpp在ros2环境下的源代码包,下载可以直接编译使用,详细用法可以参看相关文章
定义表示形状的抽象类及相应的派生类,并实现相关操作符重载。 (1)定义表示形状的抽象类Shape: 添加公有成员函数double Area(),用于计算形状面积;定义为纯虚函数; 添加公有成员函数void Show(),用于显示...
matlab期货代码EliteQuant_Cpp C / C ++高频量化交易和投资平台 平台介绍 EliteQuant是由量化交易者为量化交易者建立的永久免费开源的统一量化交易平台。 它在和上都是双重列出的。 统一一词具有两个特征。 首先,它...
由于GPU涉及到很多的图形相关技术,并不是抽象出几个简单的接口就能满足业务开发需求,因此比较主流的是一些标准化的编程接口。例如OpenGL/ES、OpenCL、vulkan、directx、cuda、metal等API。这些API可以认为是一种...
c++练习文件,算法相关的源代码,感兴趣的可以看看
正常情况下json-rpc-cpp实在linux下编译的,本代码通过修改相关的平台文件可在VS2013下可编译。
Maven 项目 springMVC_Mybatis_Axis2_Task 整合 包含 源码 相关jar 包
这是c、c++、java;相关试题大全,有面试题、笔试题、答案、面试心得等,多年心血的总结,很全
中文名: 经典Java EE企业应用实战--基于WebLogic/JBoss的JSF+EJB 3+JPA整合开发 原名: 经典Java EE企业应用实战--基于WebLogic/JBoss的JSF+EJB 3+JPA整合开发 作者: 李刚 资源格式: PDF 版本: 第一版 出版社: 电子...
主要介绍了Linux UDP socket 设置为的非阻塞模式与阻塞模式区别的相关资料,需要的朋友可以参考下
STM32F103C8T6核心板的C++空白工程,包含了改成c++版本的正点原子delay与自己写的led类,适合想在keil跑C++的开发者学习相关编写要点
java.nio包提供了支持非阻塞通信的类,主要包括: ● ServerSocketChannel:ServerSocket的替代类,支持阻塞通信与非阻塞通信。 ● SocketChannel:Socket的替代类,支持阻塞通信与非阻塞通信。 ● Selector:为...
Three.cpp 的目标是在 C++11 中完全实现库的相关部分,直到并包括修订版 65。用法使用 3pm 进行设置(需要安装 cmake 和 python) download and unzip ...
c++的基础应用 主要讲解利用MFC创建界面上的组件的相关概念和源代码 其中包括几个利用MFC创建的例子 方便大家理解