int AdminRegistryImp::addTaskReq(const TaskReq& taskReq,
                                 tars::CurrentPtr current) {

void TaskListSerial::doTask() {

int main(int argc, char *argv[]) {
    g_app.main(argc, argv);

    //初始化Server部分
    initializeServer();

  LocalRollLogger::getInstance()->setLogInfo(
      ServerConfig::Application, ServerConfig::ServerName,
      ServerConfig::LogPath, ServerConfig::LogSize, ServerConfig::LogNum,
      _communicator, ServerConfig::Log);

    g_app.waitForShutdown();

Communicator comm;

  try {
    HelloPrx prx;
    comm.stringToProxy(
        "TestApp.HelloServer.HelloObj@tcp -h 172.29.233.88 -p 20001", prx);

  /**
   * 生成代理
   * @param T
   * @param objectName
   * @param setName 指定set调用的setid
   * @param proxy
   */
  template <class T>
  void stringToProxy(const string& objectName, T& proxy,
                     const string& setName = "") {
    ServantProxy* pServantProxy = getServantProxy(objectName, setName);
    proxy = (typename T::element_type*)(pServantProxy);
  }

ServantProxy *Communicator::getServantProxy(const string &objectName,
                                            const string &setName) {
  Communicator::initialize();

void Communicator::initialize() {
  TC_LockT<TC_ThreadRecMutex> lock(*this);

  if (_initialized) return;

  _initialized = true;

void Communicator::initialize() {
......

  _servantProxyFactory = new ServantProxyFactory(this);

  //网络线程
  _clientThreadNum = TC_Common::strto<size_t>(getProperty("netthread", "1"));

  if (0 == _clientThreadNum) {
    _clientThreadNum = 1;
  } else if (MAX_CLIENT_THREAD_NUM < _clientThreadNum) {
    _clientThreadNum = MAX_CLIENT_THREAD_NUM;
  }

......

  for (size_t i = 0; i < _clientThreadNum; ++i) {
    _communicatorEpoll[i] = new CommunicatorEpoll(this, i);
    _communicatorEpoll[i]->start();
  }

CommunicatorEpoll::CommunicatorEpoll(Communicator *pCommunicator,
                                     size_t netThreadSeq)
......

  // ObjectProxyFactory 对象
  _objectProxyFactory = new ObjectProxyFactory(this);

void Communicator::initialize() {
......

  //异步线程数
  _asyncThreadNum = TC_Common::strto<size_t>(getProperty("asyncthread", "3"));

  if (_asyncThreadNum == 0) {
    _asyncThreadNum = 3;
  }

  if (_asyncThreadNum > MAX_CLIENT_ASYNCTHREAD_NUM) {
    _asyncThreadNum = MAX_CLIENT_ASYNCTHREAD_NUM;
  }

......

  for (size_t i = 0; i < _asyncThreadNum; ++i) {
    _asyncThread.push_back(new AsyncProcThread(iAsyncQueueCap, merge));
  }

const size_t MAX_CLIENT_THREAD_NUM = 64;  //客户端最大网络线程数
const size_t MAX_CLIENT_ASYNCTHREAD_NUM = 1024;  //客户端最大异步线程数
const size_t MAX_CLIENT_NOTIFYEVENT_NUM =
    2048;  //客户端每个网络线程拥有的最大通知事件的数目，客户端最大caller线程数

ServantProxy *Communicator::getServantProxy(const string &objectName,
                                            const string &setName) {
......

  return _servantProxyFactory->getServantProxy(objectName, setName);
}

  ObjectProxy** ppObjectProxy = new ObjectProxy*[_comm->getClientThreadNum()];
  assert(ppObjectProxy != NULL);

  for (size_t i = 0; i < _comm->getClientThreadNum(); ++i) {
    ppObjectProxy[i] =
        _comm->getCommunicatorEpoll(i)->getObjectProxy(name, setName);
  }

ObjectProxy *CommunicatorEpoll::getObjectProxy(const string &sObjectProxyName,
                                               const string &setName) {
  return _objectProxyFactory->getObjectProxy(sObjectProxyName, setName);
}

ObjectProxy* ObjectProxyFactory::getObjectProxy(const string& sObjectProxyName,
                                                const string& setName) {
  TC_LockT<TC_ThreadRecMutex> lock(*this);

  string tmpObjName = sObjectProxyName + ":" + setName;
  map<string, ObjectProxy*>::iterator it = _objectProxys.find(tmpObjName);
  if (it != _objectProxys.end()) {
    return it->second;
  }

  ObjectProxy* pObjectProxy =
      new ObjectProxy(_communicatorEpoll, sObjectProxyName, setName);

  _objectProxys[tmpObjName] = pObjectProxy;

  _vObjectProxys.push_back(pObjectProxy);

  _objNum++;

  return pObjectProxy;
}

ObjectProxy::ObjectProxy(CommunicatorEpoll* pCommunicatorEpoll,
                         const string& sObjectProxyName, const string& setName)
......

  _endpointManger.reset(new EndpointManager(
      this, _communicatorEpoll->getCommunicator(), _sObjectProxyName,
      _communicatorEpoll->isFirstNetThread(), _invokeSetId));

  ServantPrx sp =
      new ServantProxy(_comm, ppObjectProxy, _comm->getClientThreadNum());

ServantProxy::ServantProxy(Communicator *pCommunicator,
                           ObjectProxy **ppObjectProxy, size_t iClientThreadNum)
......

  _endpointInfo.reset(
      new EndpointManagerThread(pCommunicator, (*_objectProxy)->name()));

  for (size_t i = 0; i < _objectProxyNum; ++i) {
    (*(_objectProxy + i))->setServantProxy(this);
  }

ServantProxy *Communicator::getServantProxy(const string &objectName,
                                            const string &setName) {
.......

  return _servantProxyFactory->getServantProxy(objectName, setName);
}

  /**
   * 生成代理
   * @param T
   * @param objectName
   * @param setName 指定set调用的setid
   * @param proxy
   */
  template <class T>
  void stringToProxy(const string& objectName, T& proxy,
                     const string& setName = "") {
    ServantProxy* pServantProxy = getServantProxy(objectName, setName);
    proxy = (typename T::element_type*)(pServantProxy);
  }

  Communicator comm;

  try {
    HelloPrx prx;
    comm.stringToProxy(
        "TestApp.HelloServer.HelloObj@tcp -h 172.29.233.88 -p 20001", prx);
    comm.stringToProxy(
        "TestApp.HelloServer.HelloObj@tcp -h 172.29.233.89 -p 20001", prx);

      string sReq("hello world");
      string sRsp("");

      int iRet = prx->testHello(sReq, sRsp);

shared_ptr<tars::ResponsePacket> rep = tars_invoke(tars::TARSNORMAL,"testHello", _os, context, _mStatus);

  ObjectProxy *pObjProxy = NULL;
  ReqInfoQueue *pReqQ = NULL;

  //选择网络线程
  selectNetThreadInfo(pSptd, pObjProxy, pReqQ);

  //同步调用 new 一个ReqMonitor
  assert(msg->pMonitor == NULL);
  if (msg->eType == ReqMessage::SYNC_CALL) {
    msg->bMonitorFin = false;

    if (pSptd->_sched) {
      msg->bCoroFlag = true;
      msg->sched = pSptd->_sched;
      msg->iCoroId = pSptd->_sched->getCoroutineId();
    } else {
      msg->pMonitor = new ReqMonitor;
    }
  }

  if (!pReqQ->push_back(msg, bEmpty)) {
    TLOGERROR("[ServantProxy::invoke msgQueue push_back error num:"
              << pSptd->_netSeq << "]" << endl);

    delete msg;
    msg = NULL;

    pObjProxy->getCommunicatorEpoll()->notify(pSptd->_reqQNo, pReqQ);

    throw TarsClientQueueException("client queue full");
  }

  pObjProxy->getCommunicatorEpoll()->notify(pSptd->_reqQNo, pReqQ);

//业务线程调用, 通知对应的网络线程醒过来
// iSeq业务线程号
void CommunicatorEpoll::notify(size_t iSeq, ReqInfoQueue *msgQueue) {
  assert(iSeq < MAX_CLIENT_NOTIFYEVENT_NUM);

  if (_notify[iSeq] == NULL) {
    _notify[iSeq] = new FDInfo();
    _notify[iSeq]->iType = FDInfo::ET_C_NOTIFY;
    _notify[iSeq]->p = (void *)msgQueue;
    _notify[iSeq]->iSeq = iSeq;
    _notify[iSeq]->notify.init(&_ep);
    _notify[iSeq]->notify.add((uint64_t)_notify[iSeq]);
  } else {
    _notify[iSeq]->notify.notify();
  }
}

      if (!msg->bMonitorFin) {
        TC_ThreadLock::Lock lock(*(msg->pMonitor));

        //等待直到网络线程通知过来
        if (!msg->bMonitorFin) {
          msg->pMonitor->wait();
        }
      }

    //判断eStatus来判断状态
    assert(msg->eStatus != ReqMessage::REQ_REQ);

    //        TLOGTARS("[ServantProxy::invoke line: " << __LINE__ << " status: "
    //        << msg->eStatus << ", ret: " <<msg->response->iRet << endl);

    if (msg->adapter) {
      pSptd->_szHost = msg->adapter->endpoint().desc();
    }

    if (msg->eStatus == ReqMessage::REQ_RSP &&
        msg->response->iRet == TARSSERVERSUCCESS) {
      //成功
      return;
    }

    if (msg->eStatus == ReqMessage::REQ_TIME) {
      //超时
      delete msg;
      msg = NULL;

      throw TarsSyncCallTimeoutException(os.str());
    }

    //异常调用
    int ret = msg->response->iRet;

    delete msg;
    msg = NULL;

    TarsException::throwException(ret, os.str());

  shared_ptr<ResponsePacket> rsp = msg->response;
  // rsp = msg->response;

  delete msg;
  msg = NULL;

void CommunicatorEpoll::run() {
  ServantProxyThreadData *pSptd = ServantProxyThreadData::getData();
  assert(pSptd != NULL);
  pSptd->_netThreadSeq = (int)_netThreadSeq;

  while (!_terminate) {
    try {
      //考虑到检测超时等的情况 这里就wait100ms吧
      int num = _ep.wait(100);

      if (_terminate) break;

      //先处理epoll的网络事件
      for (int i = 0; i < num; ++i) {
        const epoll_event &ev = _ep.get(i);

        uint64_t data = TC_Epoller::getU64(ev);

        if (data == 0) continue;  // data非指针, 退出循环

        //	            int64_t ms = TNOWMS;

        handle((FDInfo *)data, ev);
      }

      //处理超时请求
      doTimeout();

      //数据上报
      doStat();
      reConnect();
    } catch (exception &e) {
      TLOGERROR("[CommunicatorEpoll:run exception:" << e.what() << "]" << endl);
    } catch (...) {
      TLOGERROR("[CommunicatorEpoll:run exception.]" << endl);
    }
  }
}

void CommunicatorEpoll::handle(FDInfo *pFDInfo, const epoll_event &ev) {
......

    } else if (FDInfo::ET_C_NOTIFY == pFDInfo->iType) {
      //队列有消息通知过来
      ReqInfoQueue *pInfoQueue = (ReqInfoQueue *)pFDInfo->p;
      ReqMessage *msg = NULL;

      size_t maxProcessCount = 0;

      try {
        while (pInfoQueue->pop_front(msg)) {
......

          try {
            msg->pObjectProxy->invoke(msg);
          } catch (exception &e) {

void ObjectProxy::invoke(ReqMessage* msg) {
......

  //选择一个远程服务的Adapter来调用
  AdapterProxy* pAdapterProxy = NULL;
  bool bFirst = _endpointManger->selectAdapterProxy(msg, pAdapterProxy);

void ObjectProxy::invoke(ReqMessage* msg) {
......
  msg->adapter = pAdapterProxy;
......
  pAdapterProxy->invoke(msg);

int AdapterProxy::invoke(ReqMessage* msg) {
......
    return invoke_connection_parallel(msg);

int AdapterProxy::invoke_connection_parallel(ReqMessage* msg) {
......
  if (_timeoutQueue->sendListEmpty()) {
    int ret = _trans->sendRequest(msg->sReqData);

      //先处理epoll的网络事件
      for (int i = 0; i < num; ++i) {
        const epoll_event &ev = _ep.get(i);

        uint64_t data = TC_Epoller::getU64(ev);

        if (data == 0) continue;  // data非指针, 退出循环

        //	            int64_t ms = TNOWMS;

        handle((FDInfo *)data, ev);
      }

      Transceiver *pTransceiver = (Transceiver *)pFDInfo->p;

      //连接出错 直接关闭连接
      if (TC_Epoller::errorEvent(ev)) {
        try {
          pTransceiver->close();
        } catch (exception &e) {
          TLOGERROR("CommunicatorEpoll::handle exp:" << e.what() << " ,line:"
                                                     << __LINE__ << endl);
        } catch (...) {
          TLOGERROR("CommunicatorEpoll::handle|" << __LINE__ << endl);
        }
        return;
      }

      //先收包
      if (TC_Epoller::readEvent(ev)) {
        try {
          handleInputImp(pTransceiver);
        } catch (exception &e) {
          TLOGERROR("CommunicatorEpoll::handle exp:" << e.what() << " ,line:"
                                                     << __LINE__ << endl);
        } catch (...) {
          TLOGERROR("CommunicatorEpoll::handle|" << __LINE__ << endl);
        }
      }

void Transceiver::finishInvoke(shared_ptr<ResponsePacket> &rsp) {
  if (_adapterProxy->endpoint().authType() == AUTH_TYPELOCAL &&
      _authState != AUTH_SUCC) {
    std::string ret(rsp->sBuffer.begin(), rsp->sBuffer.end());
    tars::AUTH_STATE tmp = AUTH_SUCC;
    tars::stoe(ret, tmp);
    tars::AUTH_STATE newstate = tmp;

    TLOGTARS("[Transceiver::finishInvoke state: " << etos(_authState) << " -> "
                                                  << etos(newstate) << endl);
    setAuthState(newstate);

    if (newstate == AUTH_SUCC) {
      // flush old buffered msg when auth is not complete
      _adapterProxy->doInvoke(true);
    } else {
      TLOGERROR("[Transceiver::finishInvoke newstate: "
                << etos(newstate) << ", error close!" << endl);
      close();
    }

    return;
  }
  _adapterProxy->finishInvoke(rsp);
}

void AdapterProxy::finishInvoke(shared_ptr<ResponsePacket>& rsp) {
  if (_objectProxy->getRootServantProxy()->tars_connection_serial() > 0) {
    finishInvoke_serial(rsp);
  } else {
    finishInvoke_parallel(rsp);
  }
}

void AdapterProxy::finishInvoke_parallel(shared_ptr<ResponsePacket>& rsp) {
......
  ReqMessage* msg = NULL;
......
    //这里的队列中的发送链表中的数据可能已经在timeout的时候删除了，因此可能会core，在erase中要加判断
    //获取请求信息
    bool retErase = _timeoutQueue->erase(rsp->iRequestId, msg);

    assert(msg->eStatus == ReqMessage::REQ_REQ);

    msg->eStatus = ReqMessage::REQ_RSP;
  }

  msg->response = rsp;

  finishInvoke(msg);

  //同步调用，唤醒ServantProxy线程
  if (msg->eType == ReqMessage::SYNC_CALL) {
    if (!msg->bCoroFlag) {
      assert(msg->pMonitor);

      TC_ThreadLock::Lock sync(*(msg->pMonitor));
      msg->bMonitorFin = true;
      msg->pMonitor->notify();
    } else {
      msg->sched->put(msg->iCoroId);
    }

    return;
  }

  //异步线程数
  _asyncThreadNum = TC_Common::strto<size_t>(getProperty("asyncthread", "3"));

  if (_asyncThreadNum == 0) {
    _asyncThreadNum = 3;
  }

  if (_asyncThreadNum > MAX_CLIENT_ASYNCTHREAD_NUM) {
    _asyncThreadNum = MAX_CLIENT_ASYNCTHREAD_NUM;
  }

class HelloCallBack : public HelloPrxCallback {
 public:
  HelloCallBack() {}

  virtual ~HelloCallBack() {}

  virtual void callback_testHello(tars::Int32 ret, const std::string& sRsp) {
    cout << "callback_testHello ret:" << ret << "|sRsp:" << sRsp << endl;
  }

  virtual void callback_testHello_exception(tars::Int32 ret) {
    cout << "callback_testHello_exception ret:" << ret << endl;
  }
};

tars_invoke_async(tars::TARSNORMAL,"testHello", _os, context, _mStatus, callback);

  if (!pReqQ->push_back(msg, bEmpty)) {
    TLOGERROR("[ServantProxy::invoke msgQueue push_back error num:"
              << pSptd->_netSeq << "]" << endl);

    delete msg;
    msg = NULL;

    pObjProxy->getCommunicatorEpoll()->notify(pSptd->_reqQNo, pReqQ);

    throw TarsClientQueueException("client queue full");
  }

  pObjProxy->getCommunicatorEpoll()->notify(pSptd->_reqQNo, pReqQ);

        //异步回调，放入回调处理线程中
        _objectProxy->getCommunicatorEpoll()->pushAsyncThreadQueue(msg);

void CommunicatorEpoll::pushAsyncThreadQueue(ReqMessage *msg) {
  _communicator->pushAsyncThreadQueue(msg);
}

void Communicator::pushAsyncThreadQueue(ReqMessage *msg) {
  if (msg->pObjectProxy->getRootServantProxy()->_callback) {
    ReqMessagePtr msgPtr = msg;

    msg->pObjectProxy->getRootServantProxy()->_callback(msgPtr);
  } else if (msg->pObjectProxy->getRootServantProxy()->_callbackHash) {
    //先不考虑每个线程队列数目不一致的情况
    _asyncThread[((uint32_t)msg->adapter->trans()->fd()) % _asyncThreadNum]
        ->push_back(msg);
  } else {
    //先不考虑每个线程队列数目不一致的情况
    _asyncThread[(_asyncSeq++) % _asyncThreadNum]->push_back(msg);
  }
}

void AsyncProcThread::push_back(ReqMessage* msg) {
  if (_merge) {
    //合并了, 直接回调
    callback(msg);
  } else {
    if (_msgQueue->size() >= _iQueueCap) {
      TLOGERROR("[AsyncProcThread::push_back] async_queue full:"
                << _msgQueue->size() << ">=" << _iQueueCap << endl);
      delete msg;
    } else {
      _msgQueue->push_back(msg);

      TC_ThreadLock::Lock lock(*this);
      notify();
    }
  }
}

void AsyncProcThread::run() {
  while (!_terminate) {
    ReqMessage* msg;

    if (_msgQueue->pop_front(msg)) {
      callback(msg);
    } else {
      TC_ThreadLock::Lock lock(*this);
      timedWait(1000);
    }
  }
}

void AsyncProcThread::callback(ReqMessage* msg) {
......
  try {
    ReqMessagePtr msgPtr = msg;
    msg->callback->dispatch(msgPtr);
  } catch (exception& e) {
    TLOGERROR("[AsyncProcThread exception]:" << e.what() << endl);
  } catch (...) {
    TLOGERROR("[AsyncProcThread exception.]" << endl);
  }
}

    g_app.main(argc, argv);
    g_app.waitForShutdown();

  /**
   * 网络线程
   * Network Thread
   */
  std::vector<NetThread *> _netThreads;

  /**
   *
   */
  vector<BindAdapterPtr> _bindAdapters;

  /**
   * 监听socket
   * Listening socket
   */
  unordered_map<int, BindAdapterPtr> _listeners;

    /**
     * Adapter所用的HandleGroup
     * the HandleGroup used by Adapter
     */
    vector<HandlePtr> _handles;

void Application::main(const string& config) {
......

    //初始化Server部分
    initializeServer();

TC_EpollServer::TC_EpollServer(unsigned int iNetThreadNum)
......

  if (_netThreadNum < 1) {
    _netThreadNum = 1;
  }

  //网络线程的配置数目不能15个
  if (_netThreadNum > 15) {
    _netThreadNum = 15;
  }

......

  for (size_t i = 0; i < _netThreadNum; ++i) {
    TC_EpollServer::NetThread *netThreads =
        new TC_EpollServer::NetThread(this, i);
    _netThreads.push_back(netThreads);
  }

void Application::main(const string& config) {
......

    vector<TC_EpollServer::BindAdapterPtr> adapters;

    //绑定对象和端口
    bindAdapter(adapters);

void Application::waitForShutdown() {
......

  _epollServer->waitForShutdown();

void TC_EpollServer::waitForShutdown() {
......

  createEpoll();

    vector<TC_EpollServer::BindAdapterPtr> adapters;

    //绑定对象和端口
    bindAdapter(adapters);

      TC_EpollServer::BindAdapterPtr bindAdapter =
          new TC_EpollServer::BindAdapter(_epollServer.get());

      _epollServer->bind(bindAdapter);

int TC_EpollServer::bind(BindAdapterPtr &lsPtr) {
  auto it = _listeners.begin();

  while (it != _listeners.end()) {
    if (it->second->getName() == lsPtr->getName()) {
      throw TC_Exception("bind name '" + lsPtr->getName() + "' conflicts.");
    }
    ++it;
  }

  const TC_Endpoint &ep = lsPtr->getEndpoint();

  TC_Socket &s = lsPtr->getSocket();

  bind(ep, s, lsPtr->isManualListen());

  _listeners[s.getfd()] = lsPtr;

  _bindAdapters.push_back(lsPtr);

  return s.getfd();
}

void TC_EpollServer::bind(const TC_Endpoint &ep, TC_Socket &s,
                          bool manualListen) {
#if TARGET_PLATFORM_WINDOWS
  int type = ep.isIPv6() ? AF_INET6 : AF_INET;
#else
  int type = ep.isUnixLocal() ? AF_LOCAL : ep.isIPv6() ? AF_INET6 : AF_INET;
#endif

#if TARGET_PLATFORM_LINUX
  if (ep.isTcp()) {
    s.createSocket(SOCK_STREAM | SOCK_CLOEXEC, type);
  } else {
    s.createSocket(SOCK_DGRAM | SOCK_CLOEXEC, type);
  }
#else
  if (ep.isTcp()) {
    s.createSocket(SOCK_STREAM, type);
  } else {
    s.createSocket(SOCK_DGRAM, type);
  }
#endif

#if TARGET_PLATFORM_WINDOWS
  s.bind(ep.getHost(), ep.getPort());
  if (ep.isTcp())
#else
  if (ep.isUnixLocal()) {
    s.bind(ep.getHost().c_str());
  } else {
    s.bind(ep.getHost(), ep.getPort());
  }
  if (ep.isTcp() && !ep.isUnixLocal())

#endif
  {
    if (!manualListen) {
      //手工监听
      s.listen(10240);
    }
    s.setKeepAlive();
    s.setTcpNoDelay();

    //不要设置close wait否则http服务回包主动关闭连接会有问题
    s.setNoCloseWait();
  }

  s.setblock(false);
}

  _listeners[s.getfd()] = lsPtr;

  _epollServer->waitForShutdown();

void TC_EpollServer::waitForShutdown() {
  if (!isMergeHandleNetThread()) startHandle();

  createEpoll();

void TC_EpollServer::createEpoll() {
  uint32_t maxAllConn = 0;

  //监听socket
  auto it = _listeners.begin();

  while (it != _listeners.end()) {
    if (it->second->getEndpoint().isTcp()) {
      //获取最大连接数
      maxAllConn += it->second->getMaxConns();

      _epoller.add(it->first, it->first, EPOLLIN);
    } else {
      maxAllConn++;
    }

    ++it;
  }

  if (maxAllConn >= (1 << 22)) {
    error("createEpoll connection num: " + TC_Common::tostr(maxAllConn) +
          " >= " + TC_Common::tostr(1 << 22));
    maxAllConn = (1 << 22) - 1;
  }

  for (size_t i = 0; i < _netThreads.size(); ++i) {
    _netThreads[i]->createEpoll(maxAllConn);
  }

  //必须先等所有网络线程调用createEpoll()，初始化list后，才能调用initUdp()
  for (size_t i = 0; i < _netThreads.size(); ++i) {
    _netThreads[i]->initUdp(_listeners);
  }
}

void TC_EpollServer::NetThread::createEpoll(uint32_t maxAllConn) {
  _list.init((uint32_t)maxAllConn, _threadIndex + 1);
}

void TC_EpollServer::waitForShutdown() {
......

  for (size_t i = 0; i < _netThreadNum; ++i) {
    _netThreads[i]->start();
  }

void Application::main(const string& config) {
......

    vector<TC_EpollServer::BindAdapterPtr> adapters;

    //绑定对象和端口
    bindAdapter(adapters);
......

      //捕捉初始化可能的异常
      try {
        TC_Common::msleep(100);
        //业务应用的初始化
        initialize();

void Application::bindAdapter(
    vector<TC_EpollServer::BindAdapterPtr>& adapters) {
......
    for (size_t i = 0; i < adapterName.size(); i++) {
      string servant =
          _conf.get("/tars/application/server/" + adapterName[i] + "<servant>");

      checkServantNameValid(servant, sPrefix);

      _servantHelper->setAdapterServant(adapterName[i], servant);

void ServantHelperManager::setAdapterServant(const string &sAdapter,
                                             const string &sServant) {
  _adapter_servant[sAdapter] = sServant;

  _servant_adapter[sServant] = sAdapter;
}

  /**
   * Adapter包含的Servant(Adapter名称:servant名称)
   */
  map<string, string> _adapter_servant;

  /**
   * Adapter包含的Servant(Servant名称:Adapter名称)
   */
  map<string, string> _servant_adapter;

void HelloServer::initialize() {
  // initialize application here:
  //...

  addServant<HelloImp>(ServerConfig::Application + "." +
                       ServerConfig::ServerName + ".HelloObj");
}

  /**
   * 添加Servant
   * @param T
   * @param id
   */
  template <typename T>
  void addServant(const string &id, Application *application,
                  bool check = false) {
    if (check && _servant_adapter.end() == _servant_adapter.find(id)) {
      cerr << "[TARS]ServantHelperManager::addServant " << id
           << " not find adapter.(maybe not conf in the web)" << endl;
      throw runtime_error("[TARS]ServantHelperManager::addServant " + id +
                          " not find adapter.(maybe not conf in the web)");
    }
    _servant_creator[id] = new ServantCreation<T>(application);
  }

/**
 * Servant
 */
template <class T>
struct ServantCreation : public ServantHelperCreation {
  ServantCreation(Application *application) : _application(application) {}
  ServantPtr create(const string &s) {
    T *p = new T;
    p->setName(s);
    p->setApplication(_application);
    return p;
  }
  Application *_application;
};

  if (_conf.getDomainVector("/tars/application/server", adapterName)) {
    for (size_t i = 0; i < adapterName.size(); i++) {
......

      bindAdapter->setHandle<ServantHandle>(
          TC_Common::strto<int>(_conf.get(sLastPath + "<threads>", "1")), this);

    /**
     * 初始化处理线程,线程将会启动
     * Initialize the processing thread, which will start
     */
    template <typename T, typename... Args>
    void setHandle(size_t n, Args &&... args) {
      if (!_handles.empty()) {
        getEpollServer()->error(
            "[BindAdapter::setHandle] handle is not empty!");
        return;
      }

      _iHandleNum = n;

      _threadDataQueue.resize(_iHandleNum + 1);
      _threadDataQueue[0] = std::make_shared<BindAdapter::DataQueue>();

      if (_pEpollServer->isMergeHandleNetThread()) {
        _iHandleNum = _pEpollServer->_netThreadNum;
      }

      for (int32_t i = 0; i < _iHandleNum; ++i) {
        HandlePtr handle = new T(args...);

        handle->setHandleIndex(i);

        handle->setEpollServer(this->getEpollServer());

        handle->setBindAdapter(this);

        _handles.push_back(handle);
      }
    }

void Application::waitForShutdown() {
......
  _epollServer->waitForShutdown();
}

void TC_EpollServer::waitForShutdown() {
  if (!isMergeHandleNetThread()) startHandle();

void TC_EpollServer::startHandle() {
  if (!this->isMergeHandleNetThread()) {
    if (!_handleStarted) {
      _handleStarted = true;

      for (auto &bindAdapter : _bindAdapters) {
        const vector<TC_EpollServer::HandlePtr> &hds =
            bindAdapter->getHandles();

        for (uint32_t i = 0; i < hds.size(); ++i) {
          if (!hds[i]->isAlive()) {
            hds[i]->start();
          }
        }
      }
    }
  }
}

void TC_EpollServer::Handle::run() {
  initialize();

  handleImp();
}

void ServantHandle::initialize() {
  ServantPtr servant =
      _application->getServantHelper()->create(_bindAdapter->getName());

  if (servant) {
    _servants[servant->getName()] = servant;
  } else {
    TLOGERROR(
        "[ServantHandle initialize createServant ret null, for adapter `" +
            _bindAdapter->getName() + "`]"
        << endl);
    cerr << "ServantHandle initialize createServant ret null, for adapter `" +
                _bindAdapter->getName() + "`]"
         << endl;

    RemoteNotify::getInstance()->report(
        "initialize createServant error: no adapter:" +
        _bindAdapter->getName());

    TC_Common::msleep(100);

    exit(-1);
  }

void TC_EpollServer::Handle::handleImp() {
  // by goodenpei, 判断是否启用顺序模式
  _bindAdapter->initThreadRecvQueue(getHandleIndex());

  startHandle();

  while (!getEpollServer()->isTerminate()) {
    //等待
    wait();

void TC_EpollServer::Handle::wait() {
  if (allAdapterIsEmpty() && allFilterIsEmpty()) {
    _bindAdapter->waitAtQueue(_handleIndex, _iWaitTime);
  }
}

void TC_EpollServer::BindAdapter::waitAtQueue(uint32_t handleIndex,
                                              uint32_t waitTime) {
  TC_ThreadLock &l = getLock(handleIndex);

  TC_ThreadLock::Lock lock(l);

  l.timedWait(waitTime);
}

void TC_EpollServer::waitForShutdown() {
......

  // 初始化 listen sockets，加入epoll管理
  // 初始化每个网络线程内的连接列表
  createEpoll();

void TC_EpollServer::createEpoll() {
......
      _epoller.add(it->first, it->first, EPOLLIN);

int TC_Epoller::add(SOCKET_TYPE fd, uint64_t data, int32_t event) {
#if TARGET_PLATFORM_IOS
  return ctrl(fd, data, event, EV_ADD | EV_ENABLE);
#else
  return ctrl(fd, data, event, EPOLL_CTL_ADD);
#endif
}

          //监听端口有请求
          if (TC_Epoller::readEvent(ev)) {
#if TARGET_PLATFORM_LINUX || TARGET_PLATFORM_IOS

            bool ret;
            do {
              ret = accept(fd, it->second->_ep.isIPv6() ? AF_INET6 : AF_INET);
            } while (ret);
#else

            accept(fd, it->second->_ep.isIPv6() ? AF_INET6 : AF_INET);
#endif
          }

  TC_Socket cs;

  cs.setOwner(false);

  //接收连接
  TC_Socket s;

  s.init(fd, false, domain);
  int iRetCode = s.accept(cs, (struct sockaddr *)stSockAddr, iSockAddrSize);

    cs.setblock(false);
    cs.setKeepAlive();
    cs.setTcpNoDelay();
    cs.setCloseWaitDefault();