简介
TcpServer实现了多线程主从Reactor服务器
结构
TcpServer相关的回调
- 连接建立回调
ConnectionCallback connectionCallback_ - 连接上数据到来时的回调
MessageCallback messageCallback_ - 连接上写完成回调
WriteCompleteCallback writeCompleteCallback_
typedefstd::function<void(constTcpConnectionPtr&)>ConnectionCallback;typedefstd::function<void(constTcpConnectionPtr&)>WriteCompleteCallback;typedefstd::function<void(constTcpConnectionPtr&,Buffer*,Timestamp)>MessageCallback;Acceptor新连接回调：
voidnewConnection(intsockfd,constInetAddress&peerAddr);TcpConnection的四个回调
voidsetConnectionCallback(constConnectionCallback&cb){connectionCallback_=cb;}/// Set message callback./// Not thread safe.voidsetMessageCallback(constMessageCallback&cb){messageCallback_=cb;}/// Set write complete callback./// Not thread safe.voidsetWriteCompleteCallback(constWriteCompleteCallback&cb){writeCompleteCallback_=cb;}voidremoveConnection(constTcpConnectionPtr&conn);事件循环线程池线程启动初始回调：
voidsetThreadInitCallback(constThreadInitCallback&cb){threadInitCallback_=cb;}Acceptor连接回调
- 从事件循环池中轮循方式选取一个
- 创建TcpConnection，将TcpServer设置的
connectionCallback_,messageCallback_和writeCompleteCallback_传递到TcpConnection，同时设置关闭连接的回调removeConnection - 创建连接建立的任务
connectEstablished放入连接事件循环的队列中异步执行
voidTcpServer::newConnection(intsockfd,constInetAddress&peerAddr){loop_->assertInLoopThread();EventLoop*ioLoop=threadPool_->getNextLoop();charbuf[64];snprintf(buf,sizeofbuf,"-%s#%d",ipPort_.c_str(),nextConnId_);++nextConnId_;string connName=name_+buf;LOG_INFO<<"TcpServer::newConnection ["<<name_<<"] - new connection ["<<connName<<"] from "<<peerAddr.toIpPort();InetAddresslocalAddr(sockets::getLocalAddr(sockfd));// FIXME poll with zero timeout to double confirm the new connection// FIXME use make_shared if necessaryTcpConnectionPtrconn(newTcpConnection(ioLoop,connName,sockfd,localAddr,peerAddr));connections_[connName]=conn;conn->setConnectionCallback(connectionCallback_);conn->setMessageCallback(messageCallback_);conn->setWriteCompleteCallback(writeCompleteCallback_);conn->setCloseCallback(std::bind(&TcpServer::removeConnection,this,_1));// FIXME: unsafeioLoop->runInLoop(std::bind(&TcpConnection::connectEstablished,conn));}连接建立回调connectEstablished
- 设置连接状态为kConnected
- 事件通道绑定连接对象，开启读事件
- 执行事件回调
connectionCallback_
voidTcpConnection::connectEstablished(){loop_->assertInLoopThread();assert(state_==kConnecting);setState(kConnected);channel_->tie(shared_from_this());channel_->enableReading();connectionCallback_(shared_from_this());}关闭连接回调
- 添加任务
removeConnectionInLoop到acceptor所在的事件循环队列中异步执行 - 连接容器中删除对应连接
- 添加任务
connectDestroyed到连接所在的事件循环队列中异步执行
voidTcpServer::removeConnection(constTcpConnectionPtr&conn){// FIXME: unsafeloop_->runInLoop(std::bind(&TcpServer::removeConnectionInLoop,this,conn));}voidTcpServer::removeConnectionInLoop(constTcpConnectionPtr&conn){loop_->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));}
