使用 ACE 库框架在 UNIX 中开发高性能并发应用

使用 ACE 库框架在 UNIX 中开发高性能并发应用
来源:developerWorks 中国 作者:Arpan Sen
　　
ACE 开放源码工具包可以帮助开发人员创建健壮的可移植多线程应用程序。本文讨论创建使用 ACE 线程的应用程序的一些方法。
Adaptive Communication Environment (ACE) 是一个高性能、开放源码、面向对象的框架和 C++ 类库，它有助于简化网络应用程序的开发。ACE 工具包包括一个操作系统层和一个封装网络 API 的 C++ 外观（facades）集合。本文讨论如何使用 ACE 线程设计高性能、并发、面向对象的网络应用程序。对 ACE 的完整说明，包括如何下载和安装这个工具包，请参见 参考资料。
用于创建和管理线程的 ACE 类
在进程中生成和管理多个线程涉及下面的类：
ACE_Thread_Manager：这是负责创建、管理和同步线程的主要的类。每种操作系统在处理线程方面有细微差异，这个类对应用程序开发人员隐藏这些差异。
ACE_Sched_Params：使用这个类管理各个线程的调度优先级，调度优先级是在 ACE 源代码发行版的 ace/Sched_Params.h 头文件中定义的。可以采用不同的调度策略，可以是 “先到先服务” 的循环方式。
ACE_TSS：在多线程应用程序中使用全局变量会导致同步问题。ACE_TSS 类提供与线程相关的存储模式，可以对那些对于程序是全局的，但是对于每个线程私有的数据提供抽象。ACE_TSS 类提供 operator() 方法，这个方法提供与线程相关的数据。

了解线程管理器类
原生操作系统线程 API 是不可移植的：存在语法和语义差异。例如，UNIX? pthread_create() 和 Windows? CreateThread() 方法都创建线程，但是语法不一样。ACE_Thread_Manager 类提供以下功能：
它可以生成一个或更多线程，每个线程运行自己指定的函数。
它可以作为一个集合（称为 线程组）管理相关的线程。
它管理各个线程的调度优先级。
它允许在线程之间进行同步。
它可以修改线程属性，比如堆栈大小。
表 1 介绍 ACE_Thread_Manager 类的重要方法。

表 1. ACE_Thread_Manager 类的方法
方法名 说明
instance ACE_Thread_Manager 类是一个单实例类，使用这个方法访问线程管理器的惟一实例。
spawn 这个方法创建一个新线程，它的一个输入参数是 C/C++ 函数指针，这个函数执行应用程序的特定工作。
exit 这个方法终止一个线程，释放这个线程的所有资源。
spawn_n 这个方法创建属于同一个线程组的多个线程。
close 这个方法关闭已经创建的所有线程并释放它们的资源。
suspend 线程管理器暂停指定的线程。
resume 线程管理器恢复执行前面暂停的线程。
使用 ACE_Thread_Manager 类的变体
可以作为单实例类使用 ACE_Thread_Manager 类，也可以创建这个类的多个实例。对于单一实例，通过调用 instance 方法访问实例。如果需要管理多个线程组，可以创建不同的线程管理器类，每个类控制它自己的线程集。
清单 1 中的示例创建一个线程。

清单 1. 使用 ACE_Thread_Manager 类创建一个线程

#include "ace/Thread_Manager.h"
#include <iostream>

void thread_start(void* arg)
{
std::cout << "Running thread..\n";
}

int ACE_TMAIN (int argc, ACE_TCHAR* argv[])
{
ACE_Thread_Manager::instance()->spawn((ACE_THR_FUNC)thread_start);
return 0;
}

清单 2 给出 spawn() 方法的原型（取自 ace/Thread_Manager.h）。

清单 2. ACE_Thread_Manager::spawn 方法的原型

int spawn (ACE_THR_FUNC func,
void *arg = 0,
long flags = THR_NEW_LWP | THR_JOINABLE | THR_INHERIT_SCHED,
ACE_thread_t *t_id = 0,
ACE_hthread_t *t_handle = 0,
long priority = ACE_DEFAULT_THREAD_PRIORITY,
int grp_id = -1,
void *stack = 0,
size_t stack_size = ACE_DEFAULT_THREAD_STACKSIZE,
const char** thr_name = 0);

对于初学者来说，创建线程需要的参数数量似乎太多了，所以我们详细讨论一下各个参数和它们的作用：
ACE_THR_FUNC func：这是在生成线程时调用的函数。
void* arg：这是在生成线程时调用的函数的参数。void* 意味着用户可以传递应用程序特有的任何数据类型，甚至可以使用某种结构把多个参数组合成单一数据。
long flags：使用 flags 变量设置生成的线程的几个属性。各个属性都由单一位表示，按照 “或” 关系组合在一起。表 2 说明一些属性。
ACE_thread_t *t_id：使用这个函数访问创建的线程的 ID。每个线程具有惟一的 ID。
long priority：这是生成的线程的优先级。
int grp_id：如果提供这个参数，那么它表示生成的线程是否属于现有的某一线程组。如果传递 -1，那么创建新的线程组并在这个组中添加生成的线程。
void* stack：这是预先分配的堆栈区域的指针。如果提供 0，就请求操作系统提供生成的线程的堆栈区域。
size_t stack_size：这个参数指定线程堆栈的大小（字节数）。如果对于前一个参数（堆栈指针）指定了 0，那么请求操作系统提供大小为 stack_size 的堆栈区域。
const char** thr_name：这个参数只与支持线程命名的平台（比如 VxWorks）相关。对于 UNIX 平台，在大多数情况下忽略它。

表 2. 线程属性及其说明
线程创建标志 说明
THR_CANCEL_DISABLE 不允许取消这个线程。
THR_CANCEL_ENABLE 允许取消这个线程。
THR_DETACHED 创建异步线程。线程的退出状态对于其他任何线程不可用。当线程终止时，操作系统回收线程资源。
THR_JOINABLE 允许新线程的退出状态对于其他线程可用。这也是 ACE 创建的线程的默认属性。当这种线程终止时，操作系统不回收线程资源，直到其他线程联结它为止。
THR_NEW_LWP 创建显式的内核级线程（而不是用户级线程）。
THR_SUSPENDED 创建处于暂停状态的新线程。
清单 3 中的示例使用线程管理器类的 spawn_n 方法创建多个线程。

清单 3. 使用 ACE_Thread_Manager 类创建多个线程

#include "ace/Thread_Manager.h"
#include <iostream>

void print (void* args)
{
int id = ACE_Thread_Manager::instance()->thr_self();
std::cout << "Thread Id: " << id << std::endl;
}

int ACE_TMAIN (int argc, ACE_TCHAR* argv[])
{
ACE_Thread_Manager::instance()->spawn_n(
4, (ACE_THR_FUNC) print, 0, THR_JOINABLE | THR_NEW_LWP);

ACE_Thread_Manager::instance()->wait();
return 0;
}

ACE 中的另一种线程创建机制
本节讨论 ACE 提供的另一种线程创建/管理机制。这种方法不需要对线程管理器进行显式的细粒度的控制。在默认情况下，每个进程在创建时有一个线程，这个线程在 main 函数开始时启动，在 main 结束时终止。其他线程都需要显式地创建。创建线程的另一种方式是创建预定义的 ACE_Task_Base 类的子类，然后覆盖 svc() 方法。新线程在 svc() 方法中启动，在 svc() 方法返回时终止。在进一步解释之前，请看一下 清单 4 所示的源代码。

清单 4. 使用 ACE_Task_Base::svc 创建线程

#include “ace/Task.h”

class Thread_1 : public ACE_Task_Base {
public:
virtual int svc( ) {
std::cout << “In child’s thread\n”;
return 0;
}
};

int main ( )
{
Thread_1 th1;
th1.activate(THR_NEW_LWP|THR_JOINABLE);
th1.wait();
return 0;
}
在 svc() 方法中编写与应用程序相关的线程行为。通过调用 activate() 方法（在 ACE_Task_Base 类中声明和定义）执行线程。在激活线程之后，main() 函数等待子线程完成执行。这就是 wait() 方法的作用：在 Thread_1 执行完之前，主线程被阻塞。这一等待过程是必需的；否则，主线程会调度子线程并执行退出。在看到主线程退出时，C 运行时会销毁所有子线程；因此，子线程可能根本没有被调度或执行。
详细了解 ACE_Task_Base 类
下面详细看看 ACE_Task_Base 中的几个方法。
清单 5 给出 activate() 方法的原型。

清单 5. ACE_Task_Base::activate 方法的原型

virtual int activate (long flags = THR_NEW_LWP | THR_JOINABLE | THR_INHERIT_SCHED,
int n_threads = 1,
int force_active = 0,
long priority = ACE_DEFAULT_THREAD_PRIORITY,
int grp_id = -1,
ACE_Task_Base *task = 0,
ACE_hthread_t thread_handles[ ] = 0,
void *stack[ ] = 0,
size_t stack_size[ ] = 0,
ACE_thread_t thread_ids[ ] = 0,
const char* thr_name[ ] = 0);

可以使用 activate() 方法创建一个或多个线程，每个线程调用相同的 svc() 方法，所有线程采用相同的优先级并具有相同的组 ID。下面简要介绍一些输入参数：
long flags：参见 表 2。
int n_threads：n_threads 指定要创建的线程的数量。
int force_active：如果这个标志是 True，而且存在这个任务已经生成的线程，那么新生成的所有线程会共享以前生成的线程的组 ID，忽略传递给 activate() 方法的值。
long priority：这个参数指定线程或线程集合的优先级。调度优先级值是与操作系统相关的，坚持使用默认值 ACE_DEFAULT_THREAD_PRIORITY 是最安全的。
ACE_hthread_t thread_handles：如果 thread_handles 不是零，那么在生成 n 个线程之后，会把各个线程句柄赋值给这个数组。
void* stack：如果指定这个参数，它指定一个指针数组，这些指针指向各个线程的堆栈基。
size_t stack_size：如果指定这个参数，它指定一个整数数组，这些整数表示各个线程堆栈的大小。
ACE_thread_t thread_ids：如果 thread_ids 不是零，那么这个参数是一个数组，其中包含 n 个新生成的线程的 ID。
清单 6 给出 ACE_Task_Base 类中另外几个有用的例程。

清单 6. ACE_Task_Base 中的其他例程

// Block the main thread until all threads of this task are completed
virtual int wait (void);

// Suspend a task
virtual int suspend (void);

// Resume a suspended task.
virtual int resume (void);

// Gets the no. of active threads within the task
size_t thread_count (void) const;

// Returns the id of the last thread whose exit caused the thread count
// of this task to 0. A zero return status implies that the result is
// unknown. Maybe no threads are scheduled.
ACE_thread_t last_thread (void) const;

为了创建处于暂停状态的线程（而不是通过调用 suspend() 方法显式地暂停），需要向 activate() 方法传递 THR_SUSPENDED 标志。可以通过调用 resume() 方法恢复执行线程，见 清单 7。

清单 7. 暂停线程和恢复执行

Thread_1 th1;
th1.activate(THR_NEW_LWP|THR_JOINABLE|THR_SUSPENDED);
…// code in the main thread
th1.resume();
…// code continues in main thread

再看看线程标志
有两种线程：内核级线程和用户级线程。如果不带任何参数调用 activate() 方法，那么默认情况下创建内核级线程。内核级线程与操作系统直接交互，由内核级调度器调度。与此相反，用户级线程在进程范围内运行，为了完成某些任务，根据需要 “分配” 内核级线程。THR_NEW_LWP 标志（activate() 方法的默认参数）总是确保新创建的线程是内核级线程。
线程钩子
ACE 提供一个全局的线程启动钩子，这允许用户执行可以应用于所有线程的任何操作。为了创建启动钩子，需要创建预定义类 ACE_Thread_Hook 的子类并提供 start() 方法定义。start() 接受两个参数：一个用户定义函数的指针和传递给这个用户定义函数的 void*。为了注册钩子，需要调用静态方法 ACE_Thread_Hook::thread_hook，见 清单 8。

清单 8. 使用全局线程钩子

#include "ace/Task.h"
#include "ace/Thread_Hook.h"
#include <iostream>

class globalHook : public ACE_Thread_Hook {
public:
virtual ACE_THR_FUNC_RETURN start (ACE_THR_FUNC func, void* arg) {
std::cout << "In created thread\n";
(*func)(arg);
}
};

class Thread_1 : public ACE_Task_Base {
public:
virtual int svc( ) {
std::cout << "In child's thread\n";
return 0;
}
};

int ACE_TMAIN (int argc, ACE_TCHAR* argv[])
{
globalHook g;
ACE_Thread_Hook::thread_hook(&g);
Thread_1 th1;
th1.activate();
th1.wait();
return 0;
}

注意，自动传递给启动钩子的 ACE_THR_FUNC 指针是在执行线程的 activate() 方法时调用的相同函数。以上代码的输出是：
In created thread
In child’s thread

结束语
本文简要讨论了如何使用 ACE 框架创建和管理线程。ACE 框架还有其他一些有用的特性，比如互斥、用于同步的保护阻塞、共享内存和网络服务。详细信息请参见 参考资料。（责任编辑：A6）