ThreadManager实现

服务器TNonblockingServer

rpc请求生成task放入到任务队列中，这个任务是Runnable的，但没有依附于某个thread，等于是封装了个run函数，供线程池工作线程调用时运行的。

void addTask(stdcxx::shared_ptr<Runnable> task) {
    threadManager_->add(task, 0LL, taskExpireTime_);
}

stdcxx::shared_ptr<Runnable> task = stdcxx::shared_ptr<Runnable>(
          new Task(processor_, inputProtocol_, outputProtocol_, this));
server_->addTask(task);

// 真正的任务，其实也是一个run函数，这个Runnable是没有thread实体的
class TNonblockingServer::TConnection::Task : public Runnable {
public:
  Task(stdcxx::shared_ptr<TProcessor> processor,
       stdcxx::shared_ptr<TProtocol> input,
       stdcxx::shared_ptr<TProtocol> output,
       TConnection* connection)
    : processor_(processor),
      input_(input),
      output_(output),
      connection_(connection),
      serverEventHandler_(connection_->getServerEventHandler()),
      connectionContext_(connection_->getConnectionContext()) {}

  void run() {
    try {
      for (;;) {
        if (serverEventHandler_) {
          serverEventHandler_->processContext(connectionContext_, connection_->getTSocket());
        }
        if (!processor_->process(input_, output_, connectionContext_)
            || !input_->getTransport()->peek()) {
          break;
        }
      }
    } catch (const TTransportException& ttx) {
      GlobalOutput.printf("TNonblockingServer: client died: %s", ttx.what());
    } catch (const std::bad_alloc&) {
      GlobalOutput("TNonblockingServer: caught bad_alloc exception.");
      exit(1);
    } catch (const std::exception& x) {
      GlobalOutput.printf("TNonblockingServer: process() exception: %s: %s",
                          typeid(x).name(),
                          x.what());
    } catch (...) {
      GlobalOutput.printf("TNonblockingServer: unknown exception while processing.");
    }

    // Signal completion back to the libevent thread via a pipe
    if (!connection_->notifyIOThread()) {
      GlobalOutput.printf("TNonblockingServer: failed to notifyIOThread, closing.");
      connection_->server_->decrementActiveProcessors();
      connection_->close();
      throw TException("TNonblockingServer::Task::run: failed write on notify pipe");
    }
  }

  TConnection* getTConnection() { return connection_; }

private:
  stdcxx::shared_ptr<TProcessor> processor_;
  stdcxx::shared_ptr<TProtocol> input_;
  stdcxx::shared_ptr<TProtocol> output_;
  TConnection* connection_;
  stdcxx::shared_ptr<TServerEventHandler> serverEventHandler_;
  void* connectionContext_;
};

线程池管理ThreadManager

线程池的管理，包括增加工作线程worker，这个worker线程是继承自Runnable的，依附于具体的线程thread，还有就是增加任务，上面说到了server是怎么调用这个增加任务的接口的。注意worker和task都是继承自runnable，一个有依附的thread，一个没有，都要实现run函数。

// 线程池里面的一个工作线程
class ThreadManager::Worker : public Runnable { // Worker是继承自Runnable的
  enum STATE { UNINITIALIZED, STARTING, STARTED, STOPPING, STOPPED };

public:
  Worker(ThreadManager::Impl* manager) : manager_(manager), state_(UNINITIALIZED) {} // 管理员也初始化进来了，可以做一些管理工作

  ~Worker() {}

private:
  bool isActive() const {
    return (manager_->workerCount_ <= manager_->workerMaxCount_)
           || (manager_->state_ == JOINING && !manager_->tasks_.empty());
  }

public:

  void run() {
    Guard g(manager_->mutex_);
    bool active = manager_->workerCount_ < manager_->workerMaxCount_;
    if (active) {
      if (++manager_->workerCount_ == manager_->workerMaxCount_) {
        manager_->workerMonitor_.notify(); // 所有工作线程启动完成通知
      }
    }

    while (active) {
      active = isActive();

      while (active && manager_->tasks_.empty()) { // 工作线程从任务队列里面取任务，没有任务的时候阻塞等待
        manager_->idleCount_++;
        manager_->monitor_.wait(); // 等待任务
        active = isActive();
        manager_->idleCount_--;
      }

      shared_ptr<ThreadManager::Task> task;

      if (active) {
        if (!manager_->tasks_.empty()) {
          task = manager_->tasks_.front(); // 取到任务了
          manager_->tasks_.pop_front();
          if (task->state_ == ThreadManager::Task::WAITING) {
            task->state_ =
                (task->getExpireTime() && task->getExpireTime() < Util::currentTime()) ?
                    ThreadManager::Task::TIMEDOUT :
                    ThreadManager::Task::EXECUTING;
          }
        }

        if (manager_->pendingTaskCountMax_ != 0
            && manager_->tasks_.size() <= manager_->pendingTaskCountMax_ - 1) {
          manager_->maxMonitor_.notify();
        }
      }

      if (task) {
        if (task->state_ == ThreadManager::Task::EXECUTING) {
          manager_->mutex_.unlock(); // 任务执行期间可以不用加任务队列的锁

          try {
            task->run();
          } catch (const std::exception& e) {
            GlobalOutput.printf("[ERROR] task->run() raised an exception: %s", e.what());
          } catch (...) {
            GlobalOutput.printf("[ERROR] task->run() raised an unknown exception");
          }

          manager_->mutex_.lock();

        } else if (manager_->expireCallback_) {
          manager_->expireCallback_(task->getRunnable());
          manager_->expiredCount_++;
        }
      }
    }

    manager_->deadWorkers_.insert(this->thread());
    if (--manager_->workerCount_ == manager_->workerMaxCount_) {
      manager_->workerMonitor_.notify();
    }
  }

private:
  ThreadManager::Impl* manager_;
  friend class ThreadManager::Impl;
  STATE state_;
};

// 工作线程里面的任务
class ThreadManager::Task : public Runnable { // Task也是继承自Runnable
public:
  enum STATE { WAITING, EXECUTING, TIMEDOUT, COMPLETE };

  Task(shared_ptr<Runnable> runnable, int64_t expiration = 0LL) // 装饰者模式
    : runnable_(runnable),
      state_(WAITING),
      expireTime_(expiration != 0LL ? Util::currentTime() + expiration : 0LL) {}

  ~Task() {}

  void run() {
    if (state_ == EXECUTING) {
      runnable_->run();
      state_ = COMPLETE;
    }
  }

  shared_ptr<Runnable> getRunnable() { return runnable_; }

  int64_t getExpireTime() const { return expireTime_; }

private:
  shared_ptr<Runnable> runnable_;
  friend class ThreadManager::Worker;
  STATE state_;
  int64_t expireTime_;
};

// 添加工作线程
// class ThreadManager::Impl : public ThreadManager，ThreadManager是抽象基类，不能直接实例化，但提供了newSimpleThreadManager静态接口，具体由Impl实现。
void ThreadManager::Impl::addWorker(size_t value) {
  std::set<shared_ptr<Thread> > newThreads;
  for (size_t ix = 0; ix < value; ix++) {
    shared_ptr<ThreadManager::Worker> worker
        = shared_ptr<ThreadManager::Worker>(new ThreadManager::Worker(this));
    newThreads.insert(threadFactory_->newThread(worker)); // 这里是工厂生产的线程，要传进去这个worker的runnable对象，这是真正的运行体
  }

  Guard g(mutex_);
  workerMaxCount_ += value;
  workers_.insert(newThreads.begin(), newThreads.end()); // 放到workers_中

  for (std::set<shared_ptr<Thread> >::iterator ix = newThreads.begin(); ix != newThreads.end();
       ++ix) {
    shared_ptr<ThreadManager::Worker> worker
        = dynamic_pointer_cast<ThreadManager::Worker, Runnable>((*ix)->runnable());
    worker->state_ = ThreadManager::Worker::STARTING;
    (*ix)->start(); // 启动工作线程，start才会真正生成线程
    idMap_.insert(std::pair<const Thread::id_t, shared_ptr<Thread> >((*ix)->getId(), *ix));
  }

  while (workerCount_ != workerMaxCount_) {
    workerMonitor_.wait(); // 等待所有工作线程初始化完成
  }
}

// 添加任务
// class ThreadManager::Impl : public ThreadManager，ThreadManager是抽象基类，不能直接实例化，但提供了newSimpleThreadManager静态接口，具体由Impl实现。
void ThreadManager::Impl::add(shared_ptr<Runnable> value, int64_t timeout, int64_t expiration) {
  Guard g(mutex_, timeout);

  if (!g) {
    throw TimedOutException();
  }

  if (state_ != ThreadManager::STARTED) {
    throw IllegalStateException(
        "ThreadManager::Impl::add ThreadManager "
        "not started");
  }

  if (pendingTaskCountMax_ > 0 && (tasks_.size() >= pendingTaskCountMax_)) {
    removeExpired(true);
  }

  if (pendingTaskCountMax_ > 0 && (tasks_.size() >= pendingTaskCountMax_)) {
    if (canSleep() && timeout >= 0) {
      while (pendingTaskCountMax_ > 0 && tasks_.size() >= pendingTaskCountMax_) {
        maxMonitor_.wait(timeout);
      }
    } else {
      throw TooManyPendingTasksException();
    }
  }

  tasks_.push_back(shared_ptr<ThreadManager::Task>(new ThreadManager::Task(value, expiration)));

  if (idleCount_ > 0) {
    monitor_.notify(); // 通知有任务加进来了
  }
}

Thread类及工厂模式

我们再次从 Java 借用了线程和一个可运行的类。 线程是实际的可调度对象。 这Runnable 是在线程内执行的逻辑。 线程实现处理所有特定于平台的线程创建和销毁问题，而 Runnable 实现处理具有特定于应用程序的每线程逻辑。 这样做的好处方法是开发人员可以轻松地继承 Runnable 类无需引入特定于平台的超类。
这里，runnable弱引用了thread，因为runnable里面不一定有thread，但thread里面一定有runnable，也解决循环引用的问题。启动线程的时候，要等到真正运行到线程入口函数。

// Thread里面包装了Runnable
class Thread {
public:
    typedef std::thread::id id_t;
    static inline bool is_current(id_t t) { return t == std::this_thread::get_id(); }
    static inline id_t get_current() { return std::this_thread::get_id(); }

    virtual ~Thread(){};
    virtual void start() = 0;
    virtual void join() = 0;
    virtual id_t getId() = 0;
    virtual stdcxx::shared_ptr<Runnable> runnable() const { return _runnable; }

protected:
    virtual void runnable(stdcxx::shared_ptr<Runnable> value) { _runnable = value; }

private:
    stdcxx::shared_ptr<Runnable> _runnable;
};

// Runnable里面可以拿到Thread，这两个是互相引用，但是Runnable是弱引用，使用的是weak_ptr，可能返回空，也就是可能没有thread，比如task就是这样
class Runnable {
public:
    virtual ~Runnable(){};
    virtual void run() = 0;
    virtual stdcxx::shared_ptr<Thread> thread() { return thread_.lock(); }
    virtual void thread(stdcxx::shared_ptr<Thread> value) { thread_ = value; }

private:
    stdcxx::weak_ptr<Thread> thread_;
};

// 由线程工厂来生产线程
stdcxx::shared_ptr<Thread> StdThreadFactory::newThread(stdcxx::shared_ptr<Runnable> runnable) const {
  stdcxx::shared_ptr<StdThread> result = stdcxx::shared_ptr<StdThread>(new StdThread(isDetached(), runnable));
  runnable->thread(result);
  return result;
}

// 真正的线程在这里
class StdThread : public Thread, public stdcxx::enable_shared_from_this<StdThread> {
public:
  enum STATE { uninitialized, starting, started, stopping, stopped };

  static void threadMain(stdcxx::shared_ptr<StdThread> thread);

private:
  std::unique_ptr<std::thread> thread_;
  Monitor monitor_;
  STATE state_;
  bool detached_;

public:
  StdThread(bool detached, stdcxx::shared_ptr<Runnable> runnable)
    : state_(uninitialized), detached_(detached) {
    this->Thread::runnable(runnable); // 设置Thread的Runnable
  }

  ~StdThread() {
    if (!detached_ && thread_->joinable()) {
      try {
        join();
      } catch (...) {
        // We're really hosed.
      }
    }
  }

  STATE getState() const
  {
    Synchronized sync(monitor_);
    return state_;
  }

  void setState(STATE newState)
  {
    Synchronized sync(monitor_);
    state_ = newState;

    // unblock start() with the knowledge that the thread has actually
    // started running, which avoids a race in detached threads.
    if (newState == started) {
	  monitor_.notify();
    }
  }

  void start() { // 调用这里才会真正生成线程
    if (getState() != uninitialized) {
      return;
    }

    stdcxx::shared_ptr<StdThread> selfRef = shared_from_this(); // 这里要获取自身的指针
    setState(starting);

    Synchronized sync(monitor_);
    thread_ = std::unique_ptr<std::thread>(new std::thread(threadMain, selfRef)); // 真正的生成线程的地方

    if (detached_)
      thread_->detach(); // 如果是detach，后面就一定需要等待了

    // Wait for the thread to start and get far enough to grab everything
    // that it needs from the calling context, thus absolving the caller
    // from being required to hold on to runnable indefinitely.
    monitor_.wait(); // 等待线程真正运行起来
  }

  void join() {
    if (!detached_ && state_ != uninitialized) {
      thread_->join();
    }
  }

  Thread::id_t getId() { return thread_.get() ? thread_->get_id() : std::thread::id(); }

  stdcxx::shared_ptr<Runnable> runnable() const { return Thread::runnable(); }

  void runnable(stdcxx::shared_ptr<Runnable> value) { Thread::runnable(value); }
};

// 线程的入口函数
void StdThread::threadMain(stdcxx::shared_ptr<StdThread> thread) {
#if GOOGLE_PERFTOOLS_REGISTER_THREAD
  ProfilerRegisterThread();
#endif

  thread->setState(started); // 线程真正运行起来了
  thread->runnable()->run(); // 线程里面的运行实体，究竟是运行什么由runnable进行包装，一般是循环处理

  if (thread->getState() != stopping && thread->getState() != stopped) { // 跑到这里说明循环完成了，线程也要结束了
    thread->setState(stopping);
  }
}