简介
TcpServer拥有Acceptor类，新连接到达时new TcpConnection后续客户端和TcpConnection类交互。TcpServer管理连接和启动线程池，用Acceptor接受连接。
服务端封装 - muduo的server端维护了多个tcpconnection
注意TcpServer本身不带Channel，而是使用Acceptor的Channel
成员及属性解析
主要接口
回调setters
这些回调函数会在新连接建立时传递给TcpConnction对象
start
启动threadPool_线程池
在runInLoop中执行acceptor的listen
这里专门设置了一个started_标记，防止多次运行start
核心实现：newConnection
从accept回调中获得的fd动态创建新的TcpConnection对象
为连接对象注册各类回调函数
将连接对象存入connectionMap_映射表里
主要成员
loop
这个loop也是acceptor的loop
acceptor
threadPool
一个EventLoopThreadPool，用来存放io线程的EventLoopThread
socket
服务端用来监听的socket
ConnectionMap
一个连接名和实例（TcpConnectionPtr）的映射容器
TcpServer中回调的传递示意简图：
源码剖析
源码已经编写注释
TcpServer.h
// Copyright 2010, Shuo Chen. All rights reserved.
// http://code.google.com/p/muduo/
//
// Use of this source code is governed by a BSD-style license
// that can be found in the License file.// Author: Shuo Chen (chenshuo at chenshuo dot com)
//
// This is a public header file, it must only include public header files.#ifndef MUDUO_NET_TCPSERVER_H
#define MUDUO_NET_TCPSERVER_H#include "muduo/base/Atomic.h"
#include "muduo/base/Types.h"
#include "muduo/net/TcpConnection.h"#include <map>namespace muduo
{
namespace net
{class Acceptor;
class EventLoop;
class EventLoopThreadPool;///
/// TCP server, supports single-threaded and thread-pool models.
///
/// This is an interface class, so don't expose too much details.
class TcpServer : noncopyable
{public:typedef std::function<void(EventLoop*)> ThreadInitCallback;enum Option{kNoReusePort,//不设置端口复用kReusePort,//设置端口复用};//TcpServer(EventLoop* loop, const InetAddress& listenAddr);TcpServer(EventLoop* loop,//mainloopconst InetAddress& listenAddr,const string& nameArg,Option option = kNoReusePort);~TcpServer(); // force out-line dtor, for std::unique_ptr members.const string& ipPort() const { return ipPort_; }const string& name() const { return name_; }EventLoop* getLoop() const { return loop_; }/// Set the number of threads for handling input.////// Always accepts new connection in loop's thread./// Must be called before @c start/// @param numThreads/// - 0 means all I/O in loop's thread, no thread will created./// this is the default value./// - 1 means all I/O in another thread./// - N means a thread pool with N threads, new connections/// are assigned on a round-robin basis.void setThreadNum(int numThreads);//设置subloop数量void setThreadInitCallback(const ThreadInitCallback& cb)//设置subloop线程初始化时调用的函数{ threadInitCallback_ = cb; }/// valid after calling start()std::shared_ptr<EventLoopThreadPool> threadPool()//返回EventLoopThread线程池{ return threadPool_; }/// Starts the server if it's not listening.////// It's harmless to call it multiple times./// Thread safe.void start();/// Set connection callback./// Not thread safe.void setConnectionCallback(const ConnectionCallback& cb){ connectionCallback_ = cb; }/// Set message callback./// Not thread safe.void setMessageCallback(const MessageCallback& cb){ messageCallback_ = cb; }/// Set write complete callback./// Not thread safe.void setWriteCompleteCallback(const WriteCompleteCallback& cb){ writeCompleteCallback_ = cb; }private:/// Not thread safe, but in loopvoid newConnection(int sockfd, const InetAddress& peerAddr);/// Thread safe.void removeConnection(const TcpConnectionPtr& conn);/// Not thread safe, but in loopvoid removeConnectionInLoop(const TcpConnectionPtr& conn);typedef std::map<string, TcpConnectionPtr> ConnectionMap;EventLoop* loop_; // the acceptor loopconst string ipPort_;const string name_;std::unique_ptr<Acceptor> acceptor_; // avoid revealing Acceptorstd::shared_ptr<EventLoopThreadPool> threadPool_;ConnectionCallback connectionCallback_;MessageCallback messageCallback_;WriteCompleteCallback writeCompleteCallback_;ThreadInitCallback threadInitCallback_;AtomicInt32 started_;// always in loop threadint nextConnId_;ConnectionMap connections_;
};} // namespace net
} // namespace muduo#endif // MUDUO_NET_TCPSERVER_HTcpServer.cc
// Copyright 2010, Shuo Chen. All rights reserved.
// http://code.google.com/p/muduo/
//
// Use of this source code is governed by a BSD-style license
// that can be found in the License file.// Author: Shuo Chen (chenshuo at chenshuo dot com)#include "muduo/net/TcpServer.h"#include "muduo/base/Logging.h"
#include "muduo/net/Acceptor.h"
#include "muduo/net/EventLoop.h"
#include "muduo/net/EventLoopThreadPool.h"
#include "muduo/net/SocketsOps.h"#include <stdio.h> // snprintfusing namespace muduo;
using namespace muduo::net;TcpServer::TcpServer(EventLoop* loop,const InetAddress& listenAddr,const string& nameArg,Option option): loop_(CHECK_NOTNULL(loop)), //TcpServer所在的主线程下运行的事件驱动循环ipPort_(listenAddr.toIpPort()),/* 服务器负责监听的本地ip和端口 */name_(nameArg),/* 服务器名字，创建时传入 */acceptor_(new Acceptor(loop, listenAddr, option == kReusePort)),/* Acceptor对象，负责监听客户端连接请求，运行在主线程的EventLoop中 */threadPool_(new EventLoopThreadPool(loop, name_)),/* 事件驱动线程池，池中每个线程运行一个EventLoop */connectionCallback_(defaultConnectionCallback),/* 用户传入，有tcp连接到达或tcp连接关闭时调用，传给TcpConnection */messageCallback_(defaultMessageCallback),/* 用户传入，对端发来消息时调用，传给TcpConnection */nextConnId_(1) /* TcpConnection特有id，每增加一个TcpConnection，nextConnId_加一 */
{ /* * 设置回调函数，当有客户端请求时，Acceptor接收客户端请求，然后调用这里设置的回调函数* 回调函数用于创建TcpConnection连接*/acceptor_->setNewConnectionCallback(std::bind(&TcpServer::newConnection, this, _1, _2));
}TcpServer::~TcpServer()
{loop_->assertInLoopThread();LOG_TRACE << "TcpServer::~TcpServer [" << name_ << "] destructing";for (auto& item : connections_){TcpConnectionPtr conn(item.second);item.second.reset();conn->getLoop()->runInLoop(std::bind(&TcpConnection::connectDestroyed, conn));}
}void TcpServer::setThreadNum(int numThreads)
{assert(0 <= numThreads);threadPool_->setThreadNum(numThreads);
}void TcpServer::start()
{if (started_.getAndSet(1) == 0){threadPool_->start(threadInitCallback_);//启动线程池,threadInitCallback_创建好所有线程后调用的回调函数assert(!acceptor_->listenning());loop_->runInLoop( //直接调用linsten函数std::bind(&Acceptor::listen, get_pointer(acceptor_)));}
}/* * Acceptor接收客户端请求后调用的回调函数* @param sockfd: 已经接收完成（三次握手完成）后的客户端套接字* @param peerAddr: 客户端地址* * Acceptor只负责接收客户端请求* TcpServer需要生成一个TcpConnection用于管理tcp连接* * 1.TcpServer内有一个EventLoopThreadPool，即事件循环线程池，池子中每个线程都是一个EventLoop* 2.每个EventLoop包含一个Poller用于监听注册到这个EventLoop上的所有Channel* 3.当建立起一个新的TcpConnection时，这个连接会放到线程池中的某个EventLoop中* 4.TcpServer中的baseLoop只用来检测客户端的连接* * 从libevent的角度看就是* 1.EventLoopThreadPool是一个struct event_base的池子，池子中全是struct event_base* 2.TcpServer独占一个event_base，这个event_base不在池子中* 3.TcpConnection会扔到这个池子中的某个event_base中*/
void TcpServer::newConnection(int sockfd, const InetAddress& peerAddr)
{loop_->assertInLoopThread();EventLoop* ioLoop = threadPool_->getNextLoop();//从事件驱动线程池中取出一个线程给TcpConnection /* 为TcpConnection生成独一无二的名字 */char buf[64];snprintf(buf, sizeof buf, "-%s#%d", ipPort_.c_str(), nextConnId_);++nextConnId_;string connName = name_ + buf;LOG_INFO << "TcpServer::newConnection [" << name_<< "] - new connection [" << connName<< "] from " << peerAddr.toIpPort();/* * 根据sockfd获取tcp连接在本地的<地址，端口>* getsockname(int fd, struct sockaddr*, int *size);*/InetAddress localAddr(sockets::getLocalAddr(sockfd));// FIXME poll with zero timeout to double confirm the new connection// FIXME use make_shared if necessary/* 创建一个新的TcpConnection代表一个Tcp连接 */TcpConnectionPtr conn(new TcpConnection(ioLoop,connName,sockfd,localAddr,peerAddr));/* 添加到所有tcp 连接的map中，键是tcp连接独特的名字（服务器名+客户端<地址，端口>） */connections_[connName] = conn;/* 为tcp连接设置回调函数（由用户提供） */conn->setConnectionCallback(connectionCallback_);conn->setMessageCallback(messageCallback_);conn->setWriteCompleteCallback(writeCompleteCallback_);/* * 关闭回调函数，由TcpServer设置，作用是将这个关闭的TcpConnection从map中删除* 当poll返回后，发现被激活的原因是EPOLLHUP，此时需要关闭tcp连接* 调用Channel的CloseCallback，进而调用TcpConnection的handleClose，进而调用removeConnection*/conn->setCloseCallback(std::bind(&TcpServer::removeConnection, this, _1)); // FIXME: unsafe/* * 连接建立后，调用TcpConnection连接建立成功的函数* 1.新建的TcpConnection所在事件循环是在事件循环线程池中的某个线程* 2.所以TcpConnection也就属于它所在的事件驱动循环所在的那个线程* 3.调用TcpConnection的函数时也就应该在自己所在线程调用* 4.所以需要调用runInLoop在自己的那个事件驱动循环所在线程调用这个函数* 5.当前线程是TcpServer的主线程，不是TcpConnection的线程，如果在这个线程直接调用会阻塞监听客户端请求* 6.其实这里不是因为线程不安全，即使在这个线程调用也不会出现线程不安全，因为TcpConnection本就是由这个线程创建的*/ioLoop->runInLoop(std::bind(&TcpConnection::connectEstablished, conn));
}void TcpServer::removeConnection(const TcpConnectionPtr& conn)
{// FIXME: unsafeloop_->runInLoop(std::bind(&TcpServer::removeConnectionInLoop, this, conn));
}void TcpServer::removeConnectionInLoop(const TcpConnectionPtr& conn)
{//关闭连接,把fd从epoll中del掉,要释放connector(包括描述符)和channelloop_->assertInLoopThread();LOG_INFO << "TcpServer::removeConnectionInLoop [" << name_<< "] - connection " << conn->name();size_t n = connections_.erase(conn->name());(void)n;assert(n == 1);EventLoop* ioLoop = conn->getLoop();ioLoop->queueInLoop(std::bind(&TcpConnection::connectDestroyed, conn));
}
设计模式)
)
