利用ace的ACE_Task等类实现线程池的方法详解

首先，需要明确线程池的概念。线程池是一种多线程处理方式，它的基本思想是在系统启动时预先创建一定数量的线程，放入线程池中，待有任务到来时就可以避免频繁地创建和销毁线程，提高系统的稳定性和效率。

接下来我将具体介绍如何利用ACE的ACE_Task等类来实现线程池。

利用ACE_Task等类实现线程池的基本原理

1.定义一个继承自ACE_Task的线程池类，并设置参数THR_DETACH表示设置线程创建为分离模式。线程池类里有一个任务队列，用于存储所有需要执行的任务。

class ThreadPool : public ACE_Task<ACE_MT_SYNCH>
{
public:
    ThreadPool(unsigned int pool_size = 5);
    virtual ~ThreadPool();

    int start();
    void stop();
    int put(ACE_Task<ACE_MT_SYNCH>* task);

protected:
    virtual int svc();

private:
    unsigned int pool_size_;
    ACE_thread_mutex task_mutex_;
    std::queue<ACE_Task<ACE_MT_SYNCH>*> task_queue_;
};

2.在构造函数中设置线程池的大小，同时创建pool_size_个线程，并将它们放入待调用队列。

ThreadPool::ThreadPool(unsigned int pool_size /*= 5*/):
pool_size_(pool_size)
{
    for (unsigned int i = 0; i < pool_size_; ++i)
    {
        ACE_Thread_Manager::instance()->spawn(this, NULL, THR_DETACH);
    }
}

3.实现put函数，将需要执行的任务塞入任务队列。同时通过task_mutex_保证线程安全。

int ThreadPool::put(ACE_Task<ACE_MT_SYNCH>* task)
{
    ACE_Guard<ACE_Thread_Mutex> guard(task_mutex_);
    task_queue_.push(task);
    return 0;
}

4.在线程池类的svc函数中通过无限循环的方式不断地取出任务队列中的任务，执行任务的run函数。

int ThreadPool::svc()
{
    for (;;)
    {
        ACE_Task<ACE_MT_SYNCH>* task = NULL;
        {
            ACE_Guard<ACE_Thread_Mutex> guard(task_mutex_);
            if (!task_queue_.empty())
            {
                task = task_queue_.front();
                task_queue_.pop();
            }
        }

        if (task)
        {
            task->open(NULL);
            task->svc();
            task->close();
        }

        ACE_OS::sleep(1);
    }
    return 0;
}

5.在析构函数里先通过stop函数停止线程池，再通过阻塞等待所有线程结束。

ThreadPool::~ThreadPool()
{
    stop();

    unsigned int size = task_queue_.size();
    for(unsigned int i = 0; i < size; ++i)
    {
        delete task_queue_.front();
        task_queue_.pop();
    }
}

void ThreadPool::stop()
{
    this->thr_mgr()->cancel_all();
}

至此，利用ACE_Task等类实现线程池的基本原理已经介绍完毕。

示例说明

1. 最简单的线程池任务，输出一个字符串

class HelloWorldTask : public ACE_Task<ACE_MT_SYNCH>
{
public:
    HelloWorldTask()
    {
    }

protected:
    virtual int svc()
    {
        ACE_DEBUG((LM_DEBUG, ACE_TEXT("Hello World\n")));
        return 0;
    }
};

我们可以在int main()函数中尝试创建线程池和任务，并调用put函数扔进任务队列：

int main(int argc, char** argv)
{
    ThreadPool tp(2);

    HelloWorldTask task1;
    HelloWorldTask task2;
    HelloWorldTask task3;
    tp.put(&task1);
    tp.put(&task2);
    tp.put(&task3);

    tp.wait();
    return 0;
}

其中wait函数用于等待线程池里的所有任务都完成。

1. 高级线程池任务，将ACE_Time_Value类中保存的毫秒转化为第几天的第几秒，封装成一个ACE_Task类。

class CustomTask : public ACE_Task<ACE_MT_SYNCH>
{
public:
    CustomTask(const ACE_Time_Value& tv):
    tv_(tv)
    {
    }

protected:
    virtual int svc()
    {
        ACE_Time_Value current_time = ACE_OS::gettimeofday();
        ACE_Time_Value diff = current_time - tv_;
        long day = diff.sec() / 86400;
        long sec = diff.sec() % 86400;
        ACE_DEBUG((LM_DEBUG, ACE_TEXT("%ld days %ld secs\n"), day, sec));
        return 0;
    }

private:
    ACE_Time_Value tv_;
};

我们可以在int main()函数中进行下列尝试：

int main(int argc, char** argv)
{
    ThreadPool tp(2);

    ACE_Time_Value tv = ACE_OS::gettimeofday();
    ACE_Time_Value delay(30000);
    CustomTask task1(tv - delay);
    CustomTask task2(tv - delay * 2);

    tp.put(&task1);
    tp.put(&task2);

    tp.wait();
    return 0;
}

结果输出的是当前时间到指定时间之间差了几天几秒。