文档章节

用 acl::master_aio 类编写高并发非阻塞服务器程序

郑树新
 郑树新
发布于 2014/09/03 12:10
字数 3257
阅读 30
收藏 0

      在文章《使用 acl::master_threads 类编写多进程多线程服务器程序》和《使用 acl::master_proc 类编写多进程服务器程序》中分别讲述了如何编写 LINUX 平台下阻塞式服务器程序的多线程和多进程方式。虽然这两种模式都可以处理并发任务,并且效率也不低,但是毕竟线程和进程资源是操作系统的宝贵资源,如果要支持非常高的并发请求,则会因为系统限制而不能创建更多的进程或线程。大家常见的 webserver Nginx 就是以支持高并发而闻名,Nginx 本身就是非阻塞设计的典型应用,当然还有很多其它服务器程序也是非阻塞的:ircd、Squid、Varnish、Bind 等。

      因为非阻塞程序编写有一定难度,所以现在有人在写非阻塞程序时不得不转向 Erlang、Nodejs 等一些脚本式语言,要想用 C/C++ 来实现非阻塞程序还是有些难度:系统级 API 过于底层、容易出错、难于调试。虽然也有一些 C++ 库(象ACE)提供了非阻塞库,但这些库的入门门槛还是挺高的。根据本人多年在非阻塞编程方面的经验,总结出一套比较实用的 C/C++ 非阻塞函数库,本文就讲述如果使用 acl_cpp 库中的 master_aio 类来编写非阻塞服务器程序(该服务器程序可以规定启动最大进程的个数,其中每个进程是一个独立的非阻塞进程)。在文章《非阻塞网络编程实例讲解》中给出了一个非阻塞网络程序的示例,您可以参考一下。

      一、类接口定义

      master_aio 是一个纯虚类,其中定义的接口需要子类实现,子类实例必须只能创建一个实例,接口如下:

 

/**
		 * 纯虚函数:当接收到一个客户端连接时调用此函数
		 * @param stream {aio_socket_stream*} 新接收到的客户端异步流对象
		 * @return {bool} 该函数如果返回 false 则通知服务器框架不再接收
		 *  远程客户端连接,否则继续接收客户端连接
		 */
		virtual bool on_accept(aio_socket_stream* stream) = 0;

 

      master_aio 提供了四个外部方法,如下:

 

/**
		 * 开始运行,调用该函数是指该服务进程是在 acl_master 服务框架
		 * 控制之下运行,一般用于生产机状态
		 * @param argc {int} 从 main 中传递的第一个参数,表示参数个数
		 * @param argv {char**} 从 main 中传递的第二个参数
		 */
		void run_daemon(int argc, char** argv);

		/**
		 * 在单独运行时的处理函数,用户可以调用此函数进行一些必要的调试工作
		 * @param addr {const char*} 服务监听地址
		 * @param conf {const char*} 配置文件全路径
		 * @param ht {aio_handle_type} 事件引擎的类型
		 * @return {bool} 监听是否成功
		 */
		bool run_alone(const char* addr, const char* conf = NULL,
			aio_handle_type ht = ENGINE_SELECT);

		/**
		 * 获得异步IO的事件引擎句柄,通过此句柄,用户可以设置定时器等功能
		 * @return {aio_handle*}
		 */
		aio_handle* get_handle() const;

		/**
		 * 在 run_alone 模式下,通知服务器框架关闭引擎,退出程序
		 */
		void stop();

 

        从上面两个函数,可以看出 master_aio 类当在生产环境下(由 acl_master 进程统一控制调度),用户需要调用 run_daemon 函数;如果用户在开发过程中需要手工进行调试,则可以调用 run_alone 函数。

        master_aio 的基类 master_base 的几个虚接口如下:

 

/**
		 * 当进程切换用户身份前调用的回调函数,可以在此函数中做一些
		 * 用户身份为 root 的权限操作
		 */
		virtual void proc_pre_jail() {}

		/**
		 * 当进程切换用户身份后调用的回调函数,此函数被调用时,进程
		 * 的权限为普通受限级别
		 */
		virtual void proc_on_init() {}

		/**
		 * 当进程退出前调用的回调函数
		 */
		virtual void proc_on_exit() {}

		// 在 run_alone 状态下运行前,调用此函数初始化一些配置

 

      基类的这几个虚函数用户可以根据需要调用。

      另外,基类 master_base 还提供了几个用来读配置选项的函数:

 

/**
		 * 设置 bool 类型的配置项
		 * @param table {master_bool_tbl*}
		 */
		void set_cfg_bool(master_bool_tbl* table);

		/**
		 * 设置 int 类型的配置项
		 * @param table {master_int_tbl*}
		 */
		void set_cfg_int(master_int_tbl* table);

		/**
		 * 设置 int64 类型的配置项
		 * @param table {master_int64_tbl*}
		 */
		void set_cfg_int64(master_int64_tbl* table);

		/**
		 * 设置 字符串 类型的配置项
		 * @param table {master_str_tbl*}
		 */
		void set_cfg_str(master_str_tbl* table);

       二、示例源程序

 

// master_aio.cpp : 定义控制台应用程序的入口点。
//

#include "stdafx.h"
#include "lib_acl.hpp"

static char *var_cfg_debug_msg;

static acl::master_str_tbl var_conf_str_tab[] = {
	{ "debug_msg", "test_msg", &var_cfg_debug_msg },

	{ 0, 0, 0 }
};

static int  var_cfg_debug_enable;
static int  var_cfg_keep_alive;
static int  var_cfg_send_banner;

static acl::master_bool_tbl var_conf_bool_tab[] = {
	{ "debug_enable", 1, &var_cfg_debug_enable },
	{ "keep_alive", 1, &var_cfg_keep_alive },
	{ "send_banner", 1, &var_cfg_send_banner },

	{ 0, 0, 0 }
};

static int  var_cfg_io_timeout;

static acl::master_int_tbl var_conf_int_tab[] = {
	{ "io_timeout", 120, &var_cfg_io_timeout, 0, 0 },

	{ 0, 0 , 0 , 0, 0 }
};

static void (*format)(const char*, ...) = acl::log::msg1;

using namespace acl;

//////////////////////////////////////////////////////////////////////////
/**
 * 延迟读回调处理类
 */
class timer_reader: public aio_timer_reader
{
public:
	timer_reader(int delay)
	{
		delay_ = delay;
		format("timer_reader init, delay: %d\r\n", delay);
	}

	~timer_reader()
	{
	}

	// aio_timer_reader 的子类必须重载 destroy 方法
	void destroy()
	{
		format("timer_reader delete, delay: %d\r\n", delay_);
		delete this;
	}

	// 重载基类回调方法
	virtual void timer_callback(unsigned int id)
	{
		format("timer_reader(%u): timer_callback, delay: %d\r\n", id, delay_);

		// 调用基类的处理过程
		aio_timer_reader::timer_callback(id);
	}

private:
	int   delay_;
};

/**
 * 延迟写回调处理类
 */
class timer_writer: public aio_timer_writer
{
public:
	timer_writer(int delay)
	{
		delay_ = delay;
		format("timer_writer init, delay: %d\r\n", delay);
	}

	~timer_writer()
	{
	}

	// aio_timer_reader 的子类必须重载 destroy 方法
	void destroy()
	{
		format("timer_writer delete, delay: %d\r\n", delay_);
		delete this;
	}

	// 重载基类回调方法
	virtual void timer_callback(unsigned int id)
	{
		format("timer_writer(%u): timer_callback, delay: %u\r\n", id, delay_);

		// 调用基类的处理过程
		aio_timer_writer::timer_callback(id);
	}

private:
	int   delay_;
};

class timer_test : public aio_timer_callback
{
public:
	timer_test() {}
	~timer_test() {}
protected:
	// 基类纯虚函数
	virtual void timer_callback(unsigned int id)
	{
		format("id: %u\r\n", id);
	}

	virtual void destroy(void)
	{
		delete this;
		format("timer delete now\r\n");
	}
private:
};
/**
 * 异步客户端流的回调类的子类
 */
class io_callback : public aio_callback
{
public:
	io_callback(aio_socket_stream* client)
		: client_(client)
		, i_(0)
	{
	}

	~io_callback()
	{
		format("delete io_callback now ...\r\n");
	}

	/**
	 * 实现父类中的虚函数,客户端流的读成功回调过程
	 * @param data {char*} 读到的数据地址
	 * @param len {int} 读到的数据长度
	 * @return {bool} 返回 true 表示继续,否则希望关闭该异步流
	 */
	virtual bool read_callback(char* data, int len)
	{
		if (++i_ < 10)
			format(">>gets(i: %d): %s\r\n", i_, data);

		// 如果远程客户端希望退出,则关闭之
		if (strncmp(data, "quit", 4) == 0)
		{
			client_->format("Bye!\r\n");
			client_->close();
		}

		// 如果远程客户端希望服务端也关闭,则中止异步事件过程
		else if (strncmp(data, "stop", 4) == 0)
		{
			client_->format("Stop now!\r\n");
			client_->close();  // 关闭远程异步流

			// 通知异步引擎关闭循环过程
			client_->get_handle().stop();
		}

		// 向远程客户端回写收到的数据

		int   delay = 0;

		if (strncmp(data, "write_delay", strlen("write_delay")) == 0)
		{
			// 延迟写过程

			const char* ptr = data + strlen("write_delay");
			delay = atoi(ptr);
			if (delay > 0)
			{
				format(">> write delay %d second ...\r\n", delay);
				timer_writer* timer = new timer_writer(delay);
				client_->write(data, len, delay * 1000000, timer);
				client_->gets(10, false);
				return true;
			}
		}
		else if (strncmp(data, "read_delay", strlen("read_delay")) == 0)
		{
			// 延迟读过程

			const char* ptr = data + strlen("read_delay");
			delay = atoi(ptr);
			if (delay > 0)
			{
				client_->write(data, len);
				format(">> read delay %d second ...\r\n", delay);
				timer_reader* timer = new timer_reader(delay);
				client_->gets(10, false, delay * 1000000, timer);
				return true;
			}
		}

		client_->write(data, len);
		//client_->gets(10, false);
		return true;
	}

	/**
	 * 实现父类中的虚函数,客户端流的写成功回调过程
	 * @return {bool} 返回 true 表示继续,否则希望关闭该异步流
	 */
	virtual bool write_callback()
	{
		return true;
	}

	/**
	 * 实现父类中的虚函数,客户端流的关闭回调过程
	 */
	virtual void close_callback()
	{
		// 必须在此处删除该动态分配的回调类对象以防止内存泄露
		delete this;
	}

	/**
	 * 实现父类中的虚函数,客户端流的超时回调过程
	 * @return {bool} 返回 true 表示继续,否则希望关闭该异步流
	 */
	virtual bool timeout_callback()
	{
		format("Timeout ...\r\n");
		return true;
	}

private:
	aio_socket_stream* client_;
	int   i_;
};

//////////////////////////////////////////////////////////////////////////

class master_aio_test : public master_aio
{
public:
	master_aio_test() { timer_test_ = new timer_test(); }

	~master_aio_test() { handle_->keep_timer(false); }

protected:
	// 基类纯虚函数:当接收到一个新的连接时调用此函数
	virtual bool on_accept(aio_socket_stream* client)
	{
		// 创建异步客户端流的回调对象并与该异步流进行绑定
		io_callback* callback = new io_callback(client);

		// 注册异步流的读回调过程
		client->add_read_callback(callback);

		// 注册异步流的写回调过程
		client->add_write_callback(callback);

		// 注册异步流的关闭回调过程
		client->add_close_callback(callback);

		// 注册异步流的超时回调过程
		client->add_timeout_callback(callback);

		// 写欢迎信息
		if (var_cfg_send_banner)
			client->format("hello, you're welcome\r\n");

		// 从异步流读一行数据
		client->gets(10, false);
		//client->read();
		return true;
	}

	// 基类虚函数:服务进程切换用户身份前调用此函数
	virtual void proc_pre_jail()
	{
		format("proc_pre_jail\r\n");
		// 只有当程序启动后才能获得异步引擎句柄
		handle_ = get_handle();
		handle_->keep_timer(true); // 允许定时器被重复触发
		// 设置第一个定时任务,每隔1秒触发一次,定时任务ID为0
		handle_->set_timer(timer_test_, 1000000, 0);
	}

	// 基类虚函数:服务进程切换用户身份后调用此函数
	virtual void proc_on_init()
	{
		format("proc init\r\n");
		// 设置第二个定时任务,每隔2秒触发一次,定时任务ID为1
		handle_->set_timer(timer_test_, 2000000, 1);
	}

	// 基类虚函数:服务进程退出前调用此函数
	virtual void proc_on_exit()
	{
		format("proc exit\r\n");
	}
private:
	timer_test* timer_test_;
	aio_handle* handle_;
};
//////////////////////////////////////////////////////////////////////////

int main(int argc, char* argv[])
{
	master_aio_test ma;  // master_aio 要求只能起一个实例

	// 设置配置参数表
	ma.set_cfg_int(var_conf_int_tab);
	ma.set_cfg_int64(NULL);
	ma.set_cfg_str(var_conf_str_tab);
	ma.set_cfg_bool(var_conf_bool_tab);

	// 开始运行

	if (argc >= 2 && strcmp(argv[1], "alone") == 0)
	{
		const char* addr = "127.0.0.1:8888";

		if (argc >= 3)
			addr = argv[2];

		format = (void (*)(const char*, ...)) printf;
		format("listen: %s now\r\n", addr);
		ma.run_alone(addr);  // 单独运行方式
	}
	else
		ma.run_daemon(argc, argv);  // acl_master 控制模式运行
	return 0;
}

 

       该示例主要实现几个功能:接收一行数据并回写该数据、延迟回写所读的数据、延迟读下一行数据、设置定时器的定时任务。因为 master_aio 类是按非阻塞模式设计的(其实,该类主要对 acl 库中的非阻塞框架库用 C++ 进行了封装),所以该例子可以支持非常大的并发请求。可以通过配置文件指定系统事件 API 选择采用 select/poll/epoll 中的一种、规定进程的空闲退出时间、预先启动子进程的数量等参数。该例子所在目录:acl_cpp/samples/master_aio。

      在这个例子中,当服务端接收到客户端连接后,非阻塞框架会通过虚函数  on_accept 回调子类处理过程,子类需要在这个虚函数中需要将一个处理 IO 过程的类实例与这个非阻塞连接流进行绑定(通过 add_xxx_callback 方式),其中处理 IO 过程的类的基类定义为:aio_callback,子类需要实现其中的虚方法。

 

      三、配置文件及程序安装

      打开 acl_cpp/samples/master_aio/aio_echo.cf 配置文件,就其中几个配置参数说明一下:

 

## 由 acl_master 用来控制服务进程池的配置项
# 为 no 表示启用该进程服务,为 yes 表示禁止该服务进程
master_disable = no

# 表示本服务器进程监听 127.0.0.1 的 8888 端口
master_service = 127.0.0.1:8888

# 表示是 TCP 套接口服务类型
master_type = inet

# 表示该服务进程池的最大进程数为 2
master_maxproc = 2

# 需要预先启动的进程数,该值不应大于 master_maxproc
master_prefork = 2

# 进程程序名
master_command = aio_echo

# 进程日志记录文件,其中 {install_path} 需要用实际的安装路径代替
master_log = {install_path}/var/log/aio_echo.log

 

# 用 select 进行循环时的时间间隔
# 单位为秒
aio_delay_sec = 1
# 单位为微秒
aio_delay_usec = 500
# 采用事件循环的方式: select(default), poll, kernel(epoll/devpoll/kqueue)
aio_event_mode = select
# 是否将 socket 接收与IO功能分开: yes/no, 如果为 yes 可以大大提高 accept() 速度
aio_accept_alone = yes
# 线程池的最大线程数, 如果该值为0则表示采用单线程非阻塞模式.
aio_max_threads = 0
# 每个线程的空闲时间.
aio_thread_idle_limit = 60
# 允许访问的客户端IP地址范围
aio_access_allow = 10.0.0.1:255.255.255.255, 127.0.0.1:127.0.0.1
# 当 acl_master 退出时,如果该值置1则该程序不等所有连接处理完毕便立即退出
aio_quick_abort = 1

 

      例如当 acl_master 服务器框架程序的安装目录为:/opt/acl,则:

      /opt/acl/libexec: 该目录存储服务器程序(acl_master 程序也存放在该目录下);

      /opt/acl/conf:该目录存放 acl_master 程序配置文件 main.cf;

      /opt/acl/conf/service:该目录存放服务子进程的程序配置文件,该路径由 main.cf 文件指定;

      /opt/acl/var/log:该目录存放日志文件;

      /opt/acl/var/pid:该目录存放进程号文件。

      该程序编译通过后,需要把可执行程序放在 /opt/acl/libexec 目录下,把配置文件放在 /opt/acl/conf/service 目录下。

      在 /opt/acl/sh 下有启动/停止 acl_master 服务进程的控制脚本;运行脚本:./start.sh,然后请用下面方法检查服务是否已经启动:

      ps -ef|grep acl_master # 查看服务器控制进程是否已经启动

      netstat -nap|grep LISTEN|grep 5001 # 查看服务端口号是否已经被监听

      当然,您也可以查看 /opt/acl/var/log/acl_master 日志文件,查看服务进程的启动过程及监听服务是否正常监听。

 

      可以命令行如下测试:telnet 127.0.0.1 8888

       原文地址

       acl_cpp 下载

       acl_cpp 的编译与使用

       更多文章

       个人微博:http://weibo.com/zsxxsz

       QQ 群:242722074

 

 

 

© 著作权归作者所有

郑树新

郑树新

粉丝 104
博文 87
码字总数 161171
作品 2
昌平
程序员
私信 提问
使用 acl 生成向导快速创建服务器程序

在以前有关使用 acl 的技术文章(如:使用 acl::masterthreads 类编写多进程多线程服务器程序 ,用 acl::masteraio 类编写高并发非阻塞服务器程序,使用 acl::master_proc 类编写多进程服务器...

郑树新
2014/09/03
0
0
acl 网络通信与服务器框架库示例列表

跨平台网络通信及服务器框架库 --- "acl" 项目里有大量的测试及应用示例,主要有三个示例集合,如下: 1、acl/samples:该目录下的例子主要是基于 libacl 及 libprotocol 两个库的例子 - 1....

郑树新
2014/07/15
0
0
使用 acl 库编写高并发非阻塞网络通信程序

一、概述 acl 库的 C 库(libacl) 的 aio 模块设计了完整的非阻塞异步 IO 通信过程,在 acl 的C++库(libaclcpp) 中封装并增强了异步通信的功能,本文主要描述了 acl C++ 库之非阻塞IO库的设计...

郑树新
2014/08/25
0
0
网络与服务器编程框架库 acl_3.0.13 发布

acl 3.0.13 版本 (项目主页:https://sourceforge.net/projects/acl/) 发布了,acl 是 one Advanced C/C++ library 的简称,主要包括网络通信库以及服务器框架库等功能,支持 Linux/Windows ...

郑树新
2013/11/04
1K
6
acl_3.4.0 发布,跨平台网络通信与服务器框架

历时一年,acl 跨平台网络通信及服务器框架库发布大版本升级。主要有以下改进: 1、重构网络协程库,支持更多的操作系统平台(Linux/FreeBSD/MacOS/Windows),且支持更多网络事件引擎:epo...

郑树新
2018/05/01
828
8

没有更多内容

加载失败,请刷新页面

加载更多

【AI实战】手把手教你深度学习文字识别(文字检测篇:基于MSER, CTPN, SegLink, EAST等方法)

文字检测是文字识别过程中的一个非常重要的环节,文字检测的主要目标是将图片中的文字区域位置检测出来,以便于进行后面的文字识别,只有找到了文本所在区域,才能对其内容进行识别。 文字检...

雪饼
今天
5
0
思维导图XMind 8 Pro 绿化方法(附序列号)

按部就班: Step 1 -全新下载最新版本的 Xmind 8(注必须是英文官方的版本,中文代{过}{滤}理网站的版本修改过,无法使用pj); Step 2 -安装完毕后,点击文末的下载按钮下载pj补丁文件包,将...

一只小青蛙
今天
10
0
数据结构(ER数据库)设计规范

表命名规范 表命名的规则分为3个层级,层级之间通过_分割,例如b_r_identity、d_l_identity。规约为: [leavel]_[type]_[name] [leavel] 表示数据库表的层级和功能,分为: s:业务无关的系统...

随风溜达的向日葵
今天
5
0
阿里Sentinel控制台源码修改-对接Apollo规则持久化

https://github.com/alibaba/Sentinel/wiki/%E5%9C%A8%E7%94%9F%E4%BA%A7%E7%8E%AF%E5%A2%83%E4%B8%AD%E4%BD%BF%E7%94%A8-Sentinel 动态规则扩展 https://github.com/alibaba/Sentinel/wiki......

jxlgzwh
昨天
7
0
在Linux系统中创建SSH服务器别名

如果你经常通过 SSH 访问许多不同的远程系统,这个技巧将为你节省一些时间。你可以通过 SSH 为频繁访问的系统创建 SSH 别名,这样你就不必记住所有不同的用户名、主机名、SSH 端口号和 IP 地...

老孟的Linux私房菜
昨天
12
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部