8.1 异步支持方案

#include <pthread.h>
#include <cstdint>
#include <list>
#include <memory>
#include <atomic>
#include <unistd.h>

//--------------------模型抽象----------------------

// 任务基类
struct ITask
{
    // 任务处理入口
    virtual void handling() {}
    ITask() : mInit(0), mTrue(0) {}

private:
    friend struct Scheduler;
    uint16_t mInit; // 初始优先级
    uint16_t mTrue; // 调度过程中的真实优先级
};
using TaskRef = std::shared_ptr<ITask>;
// 任务调度
struct Scheduler
{
    Scheduler();

    void lock() { pthread_mutex_lock(&mMutex); }
    void unlock() { pthread_mutex_unlock(&mMutex); }
    void wait() { pthread_cond_wait(&mCond, &mMutex); }
    void notify() { pthread_cond_broadcast(&mCond); }

    void stop() { mQuit = true; }
    void onstop() { mStopCnt.fetch_add(1); }
    void join()
    {
        while (true)
        {
            if (mList.empty())
                stop();
            if (mStopCnt.load() >= 3)
                break;
            notify();
           // sleep(10);
        }
    }

    void addTask(TaskRef task, uint16_t init)
    {
        // lock();
        task->mInit = init;
        task->mTrue = init;
        mList.push_back(task);
        // unlock();
    }

    bool getTask(TaskRef &task)
    {
        if (mList.empty() == false)
        {
            task = mList.front();
            mList.pop_front();

            for (auto &item : mList)
                ++item->mTrue; // 更新优先级
            return true;
        }
        return false;
    }

private:
    std::list<TaskRef> mList; // 任务链表
    volatile bool mQuit = false;
    // pthread_spinlock_t mSpin; // 调度自旋锁
    pthread_mutex_t mMutex;
    pthread_cond_t mCond;
    pthread_t threads[3];
    volatile std::atomic_int32_t mStopCnt;
    static void *worker(void *userpoint);
};
Scheduler::Scheduler()
{
    mStopCnt = 0;
    pthread_attr_t attr;

    /* Initialize mutex and condition variable objects */
    pthread_mutex_init(&mMutex, NULL);
    pthread_cond_init(&mCond, NULL);

    /* For portability, explicitly create threads in a joinable state */
    pthread_attr_init(&attr);
    pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
    pthread_create(&threads[0], &attr, worker, this);
    pthread_create(&threads[1], &attr, worker, this);
    pthread_create(&threads[2], &attr, worker, this);
}

void *Scheduler::worker(void *userpoint)
{
    TaskRef task;
    Scheduler &domain = *(Scheduler *)userpoint;
    while (true)
    {
        task.reset();
        domain.lock();
        if (domain.getTask(task) == false)
        {
            domain.wait();
        }
        domain.unlock();

        if (task.get() != nullptr)
            task->handling();

        // 判断是否销毁线程池
        if (domain.mQuit == true)
            break;
    }
    domain.onstop();
    printf("线程[%d] 退出！\n",pthread_self());
    return nullptr;
}

//-----------------------------------------测试----------------------------------

struct TaskA : public ITask
{
   TaskA(int id, int serial) : mID(id), mSerial(serial) {}
   int mID, mSerial;
   void handling() override
   {
      printf("%d: int32 serial number = %d\n", mID, mSerial);
   }
};
struct TaskB : public ITask
{
   TaskB(int id, float serial) : mID(id), mSerial(serial) {}
   int mID;
   float mSerial;
   void handling() override
   {
      printf("%d: float serial number = %f\n", mID, mSerial);
   }
};
template <typename type, typename... arg>
TaskRef maketask(arg... args)
{
   auto tsk = new type(args...);
   auto ref = std::make_shared<ITask>();
   ref.reset(tsk);
   return ref;
}
int main(int argc, char *argv[])
{
   Scheduler domain;

   int id = 0;
   while (true)
   {
      domain.lock();
      domain.addTask(maketask<TaskA>(++id, id * 10), id % 6);
      domain.addTask(maketask<TaskA>(++id, id * 100), id % 6);
      domain.addTask(maketask<TaskA>(++id, id * 1000), id % 6);
      domain.addTask(maketask<TaskB>(++id, id * 3.14159), id % 6);
      domain.addTask(maketask<TaskB>(++id, id * 18.2345), id % 6);
      domain.addTask(maketask<TaskB>(++id, id * 32.4568), id % 6);
      domain.unlock();
      domain.notify();
      if (id > 100)
         break;
   }
   // domain.stop();
   domain.join();
   printf("任务全部处理完成\n");
   return 0;
}