一：Connector

    Connector可以說是muduo庫的聯結器，負責客戶端向伺服器發起連線。實際上說白了就是封裝了socket的connect操作。

Connector類的成員如下：

class Connector : boost::noncopyable,
                  public boost::enable_shared_from_this<Connector>
{
 public:
  typedef boost::function<void (int sockfd)> NewConnectionCallback;

  Connector(EventLoop* loop, const InetAddress& serverAddr);
  ~Connector();

  void setNewConnectionCallback(const NewConnectionCallback& cb)
  { newConnectionCallback_ = cb; }

  void start();  // can be called in any thread
  void restart();  // must be called in loop thread
  void stop();  // can be called in any thread

  const InetAddress& serverAddress() const { return serverAddr_; }

 private:
  enum States { kDisconnected, kConnecting, kConnected };
  static const int kMaxRetryDelayMs = 30*1000;  //預設最大重連時間30000ms
  static const int kInitRetryDelayMs = 500;    //預設重連延遲時間500ms

  void setState(States s) { state_ = s; }
  void startInLoop();
  void stopInLoop();
  void connect();
  void connecting(int sockfd);
  void handleWrite();
  void handleError();
  void retry(int sockfd);
  int removeAndResetChannel();
  void resetChannel();

  EventLoop* loop_;               //所屬的EventLoop
  InetAddress serverAddr_;    //伺服器端的地址
  bool connect_; // atomic          
  States state_;  // FIXME: use atomic variable
  boost::scoped_ptr<Channel> channel_;    //Connector所對應的Channel
  NewConnectionCallback newConnectionCallback_;   //連線成功回撥函式
  int retryDelayMs_;    //重連延遲時間(單位ms)
};
 

建構函式是這樣的：

Connector::Connector(EventLoop* loop, const InetAddress& serverAddr)
  : loop_(loop),
    serverAddr_(serverAddr),
    connect_(false),
    state_(kDisconnected),
    retryDelayMs_(kInitRetryDelayMs)   //初始化延時
{
  LOG_DEBUG << "ctor[" << this << "]";
}

建構函式初始化了I/O執行緒，伺服器地址，並設定為未連線狀態以及初始化了重連延時時間。注意，這裡的重連是指發起連線失敗重連。

Connector啟動流程是這樣的：

//發起連線
void Connector::start()
{
  connect_ = true;
  loop_->runInLoop(boost::bind(&Connector::startInLoop, this)); // FIXME: unsafe
}

實際上呼叫了：

void Connector::startInLoop()
{
  loop_->assertInLoopThread();
  assert(state_ == kDisconnected);
  if (connect_)   //呼叫前必須connect_為true，start()函式中會這麼做
  {
    connect();  //連線具體實現
  }
  else
  {
    LOG_DEBUG << "do not connect";
  }
}
 

如果connect_為true，才真正開始連線，呼叫connect()：

//連線實現
void Connector::connect()
{
  //設定非阻塞，否則退出
  int sockfd = sockets::createNonblockingOrDie(serverAddr_.family());
  int ret = sockets::connect(sockfd, serverAddr_.getSockAddr());
  int savedErrno = (ret == 0) ? 0 : errno;
  switch (savedErrno)  //檢查錯誤碼
  {
    case 0:
    case EINPROGRESS:  //非阻塞套接字，未連線成功返回碼是EINPROGRESS表示正在連線
    case EINTR:
    case EISCONN:   //連線成功
      connecting(sockfd);
      break;

    case EAGAIN:
    case EADDRINUSE:
    case EADDRNOTAVAIL:
    case ECONNREFUSED:
    case ENETUNREACH:
      retry(sockfd);   //重連
      break;

    case EACCES:
    case EPERM:
    case EAFNOSUPPORT:
    case EALREADY:
    case EBADF:
    case EFAULT:
    case ENOTSOCK:
      LOG_SYSERR << "connect error in Connector::startInLoop " << savedErrno;
      sockets::close(sockfd);   //這幾種情況不能重連，
      break;

    default:
      LOG_SYSERR << "Unexpected error in Connector::startInLoop " << savedErrno;
      sockets::close(sockfd);
      break;
  }
}

這個函式雖然很長，但實際上就是設定套接字為非阻塞，然後底層呼叫socket的conenct()函式，進而判斷errno採取相應的操作。

操作主要有三種情況，下面分三點討論：

1.errno為EINPROGRESS、EINTR、EISCONN

    說明連線成功，呼叫connecting()函式，對成功的連線進行處理：

//如果連線成功
void Connector::connecting(int sockfd)
{
  setState(kConnecting);
  assert(!channel_);
  //Channel與sockfd關聯
  channel_.reset(new Channel(loop_, sockfd));
  //設定可寫回調函式，這時候如果socket沒有錯誤，sockfd就處於可寫狀態
  channel_->setWriteCallback(
      boost::bind(&Connector::handleWrite, this)); // FIXME: unsafe
      //設定錯誤回撥函式
  channel_->setErrorCallback(
      boost::bind(&Connector::handleError, this)); // FIXME: unsafe

  // channel_->tie(shared_from_this()); is not working,
  // as channel_ is not managed by shared_ptr
  //關注可寫事件
  channel_->enableWriting();
}

總的來說，連線成功就是更改連線狀態+設定各種回撥函式+加入poller關注可寫事件。

2.errno為 EAGAIN等錯誤碼

    說明連線暫時失敗，但是仍可能成功，需要重連。呼叫retry()函式：

//重連函式，採用back-off策略重連，也就是退避策略
//也就是重連時間逐漸延長，0.5s,1s,2s,...一直到30s
void Connector::retry(int sockfd)
{
  sockets::close(sockfd);   //先關閉連線
  setState(kDisconnected);  
  if (connect_)
  {
    LOG_INFO << "Connector::retry - Retry connecting to " << serverAddr_.toIpPort()
             << " in " << retryDelayMs_ << " milliseconds. ";
    //隔一段時間後重連，重新啟用startInLoop
    loop_->runAfter(retryDelayMs_/1000.0,
                    boost::bind(&Connector::startInLoop, shared_from_this()));
    //間隔時間2倍增長
    retryDelayMs_ = std::min(retryDelayMs_ * 2, kMaxRetryDelayMs);
  }
  else   //超出最大重連時間後，輸出連線失敗
  {
    LOG_DEBUG << "do not connect";
  }
}

3.徹底失敗，返回errno為EACCES等錯誤碼

    這種情況只能關掉sockfd，因為再怎麼試也成功不了的。

 sockets::close(sockfd);

斷開連線函式就不剖析了，相對比較簡單，接下來看TcpClient類。

二：TcpClient

Tcpclient主要有下列成員：

 EventLoop* loop_;
  ConnectorPtr connector_; // avoid revealing Connector
  const string name_;
  ConnectionCallback connectionCallback_;
  MessageCallback messageCallback_;
  WriteCompleteCallback writeCompleteCallback_;
  bool retry_;   // atomic   //是否重連，是指建立的連線成功後又斷開是否重連。而Connector的重連是一直不成功是否重試的意思
  bool connect_; // atomic
  // always in loop thread
  int nextConnId_;  //name_+nextConnid_用於標識一個連線
  mutable MutexLock mutex_;
  TcpConnectionPtr connection_; // @GuardedBy mutex_ //Connector連線成功後，得到一個TcpConnection

需要注意的是，這裡的重連是連線成功後又斷開的重連，而不是Connector類的連不上一直嘗試的重連。

建構函式：

TcpClient::TcpClient(EventLoop* loop,
                     const InetAddress& serverAddr,
                     const string& nameArg)
  : loop_(CHECK_NOTNULL(loop)),
    connector_(new Connector(loop, serverAddr)),
    name_(nameArg),
    connectionCallback_(defaultConnectionCallback),
    messageCallback_(defaultMessageCallback),
    retry_(false),
    connect_(true),
    nextConnId_(1)
{
  //一旦連線建立連線，回撥newConnection
  connector_->setNewConnectionCallback(
      boost::bind(&TcpClient::newConnection, this, _1));
}

建構函式建立了一個Connector，畢竟是TcpClient嘛，需要一個東西來發起連線。並且註冊連線成功的回撥函式。

它的使用方法，先connect：

void TcpClient::connect()
{
  connect_ = true;
  connector_->start();
}

底層就是呼叫我們上文中分析的Connector的一系列連線機制，向服務端發起連線。

連線成功後，就會呼叫自己的成員函式newConnection()函式：

void TcpClient::newConnection(int sockfd)
{
  loop_->assertInLoopThread();
  InetAddress peerAddr(sockets::getPeerAddr(sockfd));
  char buf[32];
  snprintf(buf, sizeof buf, ":%s#%d", peerAddr.toIpPort().c_str(), nextConnId_);
  ++nextConnId_;
  string connName = name_ + buf;

  InetAddress localAddr(sockets::getLocalAddr(sockfd));
  // FIXME poll with zero timeout to double confirm the new connection
  // FIXME use make_shared if necessary
  //建立一個TcpConnection物件，智慧指標。
  //根據Connector中的handleWrite()函式，連線建立後會把sockfd從poller中移除，以後不會再關注可寫事件了
  //否則會出現busy loop，因為已連線套接字一直處於可寫狀態
  TcpConnectionPtr conn(new TcpConnection(loop_,
                                          connName,
                                          sockfd,
                                          localAddr,
                                          peerAddr));
  
  conn->setConnectionCallback(connectionCallback_);
  conn->setMessageCallback(messageCallback_);
  conn->setWriteCompleteCallback(writeCompleteCallback_);
  conn->setCloseCallback(
      boost::bind(&TcpClient::removeConnection, this, _1)); // FIXME: unsafe
  {
    MutexLockGuard lock(mutex_);
    connection_ = conn;   //儲存TcpConnection
  }
  conn->connectEstablished();  //這裡會關注可讀事件，並且回撥connectionCallback_，
}

該函式建立一個堆上區域性TcpConnection物件，並用TcpClient的智慧指標connection_儲存起來，這樣本函式中conn即便析構掉，connection_依然維護該連線。

然後設定各種回撥函式。由於為了避免busy loop，在Connector中一旦連線成功，我們取消關注sockfd的可寫事件。並且本函式使用conn->connectEstablished()內部會關注可讀事件：

void TcpConnection::connectEstablished()
{
  loop_->assertInLoopThread();   //斷言處於loop執行緒
  assert(state_ == kConnecting);   //斷言處於未連線狀態
  setState(kConnected);   //將狀態設定為已連線

  channel_->tie(shared_from_this());   //將自身這個TcpConnection物件提升，由於是智慧指標，所以不能直接用this
  //shared_from_this()之後引用計數+1，為3，但是shared_from_this()是臨時物件，析構後又會減一，
  //而tie是weak_ptr並不會改變引用計數，所以該函式執行完之後引用計數不會更改
  
  channel_->enableReading();   //一旦連線成功就關注它的可讀事件，加入到Poller中關注

  connectionCallback_(shared_from_this());
}

下面是連線斷開的函式：

//連線斷開
void TcpClient::removeConnection(const TcpConnectionPtr& conn)
{
  loop_->assertInLoopThread();
  assert(loop_ == conn->getLoop());

  {
    MutexLockGuard lock(mutex_);
    assert(connection_ == conn);
    connection_.reset();   //重置
  }

  //I/O執行緒中銷燬
  loop_->queueInLoop(boost::bind(&TcpConnection::connectDestroyed, conn));
  if (retry_ && connect_)  //是否發起重連
  {
    LOG_INFO << "TcpClient::connect[" << name_ << "] - Reconnecting to "
             << connector_->serverAddress().toIpPort();
    //這裡的重連是連線成功後斷開的重連，所以實際上是重啟
    connector_->restart();
  }
}

基本就這些了。