2. 程式人生 > >#muduo學習筆記(二)Reactor關鍵結構

#muduo學習筆記(二)Reactor關鍵結構

阿新 • • 發佈：2018-10-31

目錄

muduo學習筆記(二)Reactor關鍵結構

Reactor簡述

什麼是Reactor

Reactor是一種基於事件驅動的設計模式，即通過回撥機制,我們將事件的介面註冊到Reactor上,當事件發生之後,就會回撥註冊的介面。
Reactor必要的幾個元件：
Event Multiplexer事件分發器:即一些I/O複用機制select、poll、epoll等.程式將事件源註冊到分發器上,等待事件的觸發，做相應處理.
Handle事件源：用於標識一個事件，Linux上是檔案描述符.
Reactor反應器：用於管理事件的排程及註冊刪除.當有啟用的事件時,則呼叫回撥函式處理，沒有則繼續事件迴圈.
event handler事件處理器：管理已註冊事件和的排程，分成不同型別的事件(讀/寫,定時)當事件發生,呼叫對應的回撥函式處理.

Reactor模型的優缺點

優點
1）響應快，不必為單個同步時間所阻塞，雖然Reactor本身依然是同步的；
2）程式設計相對簡單，可以最大程度的避免複雜的多執行緒及同步問題，並且避免了多執行緒/程序的切換開銷；
3）可擴充套件性，可以方便的通過增加Reactor例項個數來充分利用CPU資源；
4）可複用性，reactor框架本身與具體事件處理邏輯無關，具有很高的複用性；

缺點
Reactor模式在IO讀寫資料時還是在同一個執行緒中實現的，即使使用多個Reactor機制的情況下，那些共享一個Reactor的Channel如果出現一個長時間的資料讀寫，會影響這個Reactor中其他Channel的相應時間，比如在大檔案傳輸時，IO操作就會影響其他Client的相應時間，因而對這種操作，使用傳統的Thread-Per-Connection或許是一個更好的選擇，或則此時使用Proactor模式。

poll簡述

poll的使用方法與select相似，輪詢多個檔案描述符，有讀寫時設定相應的狀態位，poll相比select優在沒有最大檔案描述符數量的限制.
# include <poll.h>
int poll ( struct pollfd * fds, unsigned int nfds, int timeout);

struct pollfd {
int fd;         /* 檔案描述符 */
short events;         /* 等待的事件 */
short revents;       /* 實際發生了的事件 */
} ;

　　每一個pollfd結構體指定了一個被監視的檔案描述符，可以傳遞多個結構體，指示poll()監視多個檔案描述符。每個結構體的events域是監視該檔案描述符的事件掩碼，由使用者來設定這個域。revents域是檔案描述符的操作結果事件掩碼，核心在呼叫返回時設定這個域。events域中請求的任何事件都可能在revents域中返回。合法的事件如下：

　　POLLIN 　　　　　　　　有資料可讀。
　　POLLRDNORM 　　　　 有普通資料可讀。
　　POLLRDBAND　　　　　 有優先資料可讀。
　　POLLPRI　　　　　　　　 有緊迫資料可讀。
　　POLLOUT　　　　　　 寫資料不會導致阻塞。
　　POLLWRNORM　　　　　 寫普通資料不會導致阻塞。
　　POLLWRBAND　　　　　 寫優先資料不會導致阻塞。
　　POLLMSGSIGPOLL 　　　　訊息可用。

poll使用樣例

#include <fcntl.h>
#include <stdio.h>
#include <unistd.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <errno.h>
#include <poll.h>

#define MAX_BUFFER_SIZE 1024
#define IN_FILES 1
#define MAX(a,b) ((a>b)?(a):(b))

int main(int argc ,char **argv)
{
  struct pollfd fds[3];
  char buf[1024];
  int i,res,real_read, maxfd;

  if((fds[0].fd=open("/dev/stdin",O_RDONLY|O_NONBLOCK)) < 0)
  {
    fprintf(stderr,"open data1 error:%s",strerror(errno));
    return 1;
  }

  for (i = 0; i < IN_FILES; i++)
  {
    fds[i].events = POLLIN | POLLPRI;
  }

  while(1) //|| fds[1].events || fds[2].events)
  {
    int ret = poll(fds, 1, 1000);
    if (ret < 0)
    {
      printf("Poll error : %s\n",strerror(errno));
      return 1;
    }

    if(ret == 0){
      printf("Poll timeout\n");
      continue;
    }

    for (i = 0; i< 1; i++)
    {
      if (fds[i].revents)
      {
        memset(buf, 0, MAX_BUFFER_SIZE);
        real_read = read(fds[i].fd, buf, MAX_BUFFER_SIZE);
        if (real_read < 0)
        {
          if (errno != EAGAIN)
          {
            printf("read eror : %s\n",strerror(errno));
            continue;
          }
        }
        else if (!real_read)
        {
          close(fds[i].fd);
          fds[i].events = 0;
        }
        else
        {
          if (i == 0)
          {
            buf[real_read] = '\0';
            printf("%s", buf);
            if ((buf[0] == 'q') || (buf[0] == 'Q'))
            {
              printf("quit\n");
              return 1;
            }
          }
          else
          {
            buf[real_read] = '\0';
            printf("%s", buf);
          }
        }
      }
    }
  }

  exit(0);
}

muduo Reactor關鍵結構

muduo Reactor最核心的事件分發機制， 即將IO multiplexing拿到的IO事件分發給各個檔案描述符（fd）的事件處理函式。

Channel

Chanel目前我對它的理解是，它負責管理一個檔案描述符(file descript)IO事件.
Channel會封裝C的poll，把不同的IO事件分發到不同的回撥：ReadCallBack、WriteCallBack等
每個Channel物件自始至終只屬於一個EventLoop，因此每個Channel物件都只屬於某一個IO執行緒。 每個Channel物件自始至終只負責一個檔案描述符（fd） 的IO事件分發

#ifndef NET_CHANNEL_H
#define NET_CHANNEL_H

#include <functional>

#include "EventLoop.hh"

class Channel {
public:
  typedef std::function<void()> EventCallBack;
  Channel(EventLoop* loop, int fd);
  ~Channel();

  void handleEvent();
  void setReadCallBack(const EventCallBack& cb) { m_readCallBack = cb; }
  void setWriteCallBack(const EventCallBack& cb) { m_writeCallBack = cb; }
  void setErrorCallBack(const EventCallBack& cb) { m_errorCallBack = cb; }

  int fd() const { return m_fd; }
  int events() const { return m_events; }
  void set_revents(int revt) { m_revents = revt; }
  bool isNoneEvent() const { return m_events == kNoneEvent; }

  void eableReading() { m_events |=  kReadEvent; update(); }

  int index() { return m_index; }
  void set_index(int idx) { m_index =idx; }

  EventLoop* ownerLoop() { return m_pLoop; }

private:
  Channel& operator=(const Channel&);
  Channel(const Channel&);

  void update();

  static const int kNoneEvent;
  static const int kReadEvent;
  static const int kWriteEvent;

  EventLoop* m_pLoop;
  const int m_fd;
  int m_events;    // 等待的事件
  int m_revents;   // 實際發生了的事件
  int m_index;

  EventCallBack m_readCallBack;
  EventCallBack m_writeCallBack;
  EventCallBack m_errorCallBack;
};

#endif

//Channel.cpp

#include <poll.h>
#include "Channel.hh"
#include "Logger.hh"

const int Channel::kNoneEvent = 0;
const int Channel::kReadEvent = POLLIN | POLLPRI;
const int Channel::kWriteEvent = POLLOUT;

Channel::Channel(EventLoop* loop, int fd)
  : m_pLoop(loop),
    m_fd(fd),
    m_events(0),
    m_revents(0),
    m_index(-1)
{

}

Channel::~Channel()
{

}

void Channel::update()
{
  m_pLoop->updateChannel(this);
}


void Channel::handleEvent()
{
  if(m_revents & POLLNVAL)
  {
    LOG_WARN << "Channel::handleEvent() POLLNVAL";
  }

  if(m_revents & (POLLERR | POLLNVAL)){
    if(m_errorCallBack) m_errorCallBack();
  }

  if(m_revents & (POLLIN | POLLPRI | POLLRDHUP)){
    if(m_readCallBack) m_readCallBack();
  }

  if(m_revents & POLLOUT){
    if(m_writeCallBack) m_writeCallBack();
  }

}

值得一提的就是 Channel::update()它會呼叫EventLoop::updateChannel()， 後者會轉而調
用Poller::updateChannel()。Poller物件下面會講，通過Poller::updateChannel()註冊IO事件(即file descript).

Channel::handleEvent()是Channel的核心， 它由EventLoop::loop()調
用， 它的功能是根據revents發生事件的的值分別呼叫不同的使用者回撥。 這個函式以後還會擴充。

Poller

Poller class是IO multiplexing的封裝。 它現在是個具體類，而在muduo中是個抽象基類，因為muduo同時支援poll(2)和epoll(4)兩種IOmultiplexing機制。
Poller是EventLoop的間接成員，只供其自己在EventLoop的IO執行緒中呼叫，因此無須加鎖。其生命期與EventLoop相等。
Poller並不擁有管理檔案描述符事件的Channel， Channel在析構之前必須自己
unregister（EventLoop::removeChannel()） ， 避免空懸指標

#ifndef _NET_POLLER_HH
#define _NET_POLLER_HH

#include <vector>
#include <map>

#include "TimeStamp.hh"
#include "EventLoop.hh"
#include "Channel.hh"

struct pollfd;

class Poller{
public:
  typedef std::vector<Channel*> ChannelList;

  Poller(EventLoop* loop);
  ~Poller();

  TimeStamp poll(int timeoutMs, ChannelList* activeChannels);

  void updateChannel(Channel* channel);

  void assertInLoopThread() { m_pOwerLoop->assertInLoopThread(); }

private:
  Poller& operator=(const Poller&);
  Poller(const Poller&);

  void fillActiveChannels(int numEvents, ChannelList* activeChannels) const;

  typedef std::vector<struct pollfd> PollFdList;
  typedef std::map<int, Channel*> ChannelMap;

  EventLoop* m_pOwerLoop;
  PollFdList m_pollfds;
  ChannelMap m_channels;

};

#endif

//Poller.cpp
#include "Poller.hh"
#include "Logger.hh"
#include <assert.h>
#include <poll.h>
#include <signal.h>

Poller::Poller(EventLoop* loop)
  : m_pOwerLoop(loop)
{

}

Poller::~Poller()
{

}

TimeStamp Poller::poll(int timeoutMs, ChannelList* activeChannels)
{
  LOG_TRACE << "Poller::poll()";
  int numEvents = ::poll(/*&*m_pollfds.begin()*/m_pollfds.data(), m_pollfds.size(), timeoutMs);
  TimeStamp now(TimeStamp::now());
  if(numEvents > 0){
    LOG_TRACE << numEvents << " events happended";
    fillActiveChannels(numEvents, activeChannels);
  }
  else if(numEvents == 0){
    LOG_TRACE << " nothing happended";
  }
  else{
    LOG_SYSERR << "Poller::poll()";
  }

  return now;
}

/*
 *fillActiveChannels()遍歷m_pollfds， 找出有活動事件的fd， 把它對應
 *的Channel填入activeChannels。
 */

void Poller::fillActiveChannels(int numEvents, ChannelList* activeChannels) const
{
  for(PollFdList::const_iterator pfd = m_pollfds.begin();
      pfd != m_pollfds.end() && numEvents > 0; ++pfd)
  {
    if(pfd->revents > 0)
    {
      --numEvents;
      ChannelMap::const_iterator ch = m_channels.find(pfd->fd);
      assert(ch != m_channels.end());
      Channel* channel = ch->second;
      assert(channel->fd() == pfd->fd);
      channel->set_revents(pfd->revents);
      activeChannels->push_back(channel);
    }
  }
}

void Poller::updateChannel(Channel* channel)
{
  assertInLoopThread();
  LOG_TRACE << "fd= " << channel->fd() << " events" << channel->events();
  if(channel->index() < 0){
    //a new one , add to pollfds
    assert(m_channels.find(channel->fd()) == m_channels.end());
    struct pollfd pfd;
    pfd.fd = channel->fd();
    pfd.events = static_cast<short>(channel->events());
    pfd.revents = 0;
    m_pollfds.push_back(pfd);
    int idx = static_cast<int>(m_pollfds.size()) - 1;
    channel->set_index(idx);
    m_channels[pfd.fd] = channel;

  }
  else{
    //update existing one
    assert(m_channels.find(channel->fd()) != m_channels.end());
    assert(m_channels[channel->fd()] == channel);
    int idx = channel->index();
    assert(0 <= idx && idx < static_cast<int>(m_pollfds.size()));
    struct pollfd& pfd = m_pollfds[idx];
    assert(pfd.fd == channel->fd() || pfd.fd == -1);
    pfd.events = static_cast<short>(channel->events());
    pfd.revents = 0;
    if(channel->isNoneEvent()){
      //ignore this pollfd
      pfd.fd = -1;
    }
  }

}

EventLoop

EventLopp在上一篇文章寫過,這裡給出改動.

EventLoop 新增了quit()成員函式， 還加了幾個資料成員，並在建構函式裡初始化它們。注意EventLoop通過智慧指標來間接持有poller.

+class Poller;
+class Channel;

class EventLoop
------------
    bool isInloopThread() const {return m_threadId == CurrentThread::tid(); }

    +void quit();
    +void updateChannel(Channel* channel);

    static EventLoop* getEventLoopOfCurrentThread();

private:
    EventLoop& operator=(const EventLoop&);
    EventLoop(const EventLoop&);

    void abortNotInLoopThread();

    +typedef std::vector<Channel*> ChannelList;

    bool m_looping;
    +bool m_quit;
    const pid_t m_threadId;
    +std::unique_ptr<Poller> m_poller;
    +ChannelList m_activeChannels;
};

//EventLoop.cpp
  m_threadId(CurrentThread::tid()),
 + m_poller(new Poller(this))
{
------
+void EventLoop::quit()
+{
+  m_quit = true;
+  //wakeup();
+}
+
+void EventLoop::updateChannel(Channel* channel)
+{
+  assert(channel->ownerLoop() == this);
+  assertInLoopThread();
+  m_poller->updateChannel(channel);
+}

上一篇文章的EventLoop->loop()什麼也沒做,現在它有了實實在在的使命，它呼叫Poller::poll()獲得當前活動事件的Chanel列表， 然後依次呼叫每個Channel的handleEvent()函式

void EventLoop::loop()
{
  assert(!m_looping);
  assertInLoopThread();
  m_looping = true;
  m_quit = false;

  LOG_TRACE << "EventLoop " << this << " start loopig";

  while(!m_quit)
  {
    m_activeChannels.clear();
    m_poller->poll(1000, &m_activeChannels);
    for(ChannelList::iterator it = m_activeChannels.begin();
      it != m_activeChannels.end(); ++it)
    {
      (*it)->handleEvent();
    }

  }

  LOG_TRACE << "EventLoop " << this << " stop loopig";
  m_looping = false;

}

Reactor時序圖

測試程式-單次觸發的定時器

程式利用timerfd_create 把時間變成了一個檔案描述符，該“檔案”在定時器超時的那一刻變得可讀，這樣就能很方便地融入到 select/poll 框架中，用統一的方式來處理 IO 事件和超時事件，這也正是 Reactor 模式的長處。

#include <errno.h>
#include <thread>
#include <strings.h>

#include "EventLoop.hh"
#include "Channel.hh"
#include "Poller.hh"

//Reactor Test
//單次觸發定時器
#include <sys/timerfd.h>

EventLoop* g_loop;

void timeout()
{
  printf("timeout!\n");
  g_loop->quit();
}

int main()
{

  EventLoop loop;
  g_loop = &loop;

  int timerfd = ::timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK |TFD_CLOEXEC);

  Channel channel(&loop, timerfd);
  channel.setReadCallBack(timeout);
  channel.eableReading();

  struct itimerspec howlong;
  bzero(&howlong, sizeof howlong);
  howlong.it_value.tv_sec = 3;
  timerfd_settime(timerfd, 0, &howlong, NULL);

  loop.loop();

  close(timerfd);

}

 ./test.out 
2018-10-31 22:25:54.532487 [TRACE] [EventLoop.cpp:16] [EventLoop] EventLoop Create 0x7FFEB9567CC0 in thread 3075
2018-10-31 22:25:54.533563 [TRACE] [Poller.cpp:64] [updateChannel] fd= 3 events3
2018-10-31 22:25:54.534000 [TRACE] [EventLoop.cpp:41] [loop] EventLoop 0x7FFEB9567CC0 start loopig
2018-10-31 22:25:54.534334 [TRACE] [Poller.cpp:20] [poll] Poller::poll()
2018-10-31 22:25:55.535827 [TRACE] [Poller.cpp:28] [poll]  nothing happended
2018-10-31 22:25:55.536287 [TRACE] [Poller.cpp:20] [poll] Poller::poll()
2018-10-31 22:25:56.538334 [TRACE] [Poller.cpp:28] [poll]  nothing happended
2018-10-31 22:25:56.538802 [TRACE] [Poller.cpp:20] [poll] Poller::poll()
2018-10-31 22:25:57.534175 [TRACE] [Poller.cpp:24] [poll] 1 events happended
timeout!
2018-10-31 22:25:57.534766 [TRACE] [EventLoop.cpp:55] [loop] EventLoop 0x7FFEB9567CC0 stop loopig

相關推薦

#muduo學習筆記(二)Reactor關鍵結構

目錄 muduo學習筆記(二)Reactor關鍵結構 Reactor簡述 什麼是Reactor Reactor模型的優缺點 poll簡述 poll使用樣例 muduo Reactor關鍵結構

muduo學習筆記(二)Reactor關鍵結構

Reactor簡述 什麼是Reactor Reactor是一種基於事件驅動的設計模式，即通過回撥機制,我們將事件的介面註冊到Reactor上,當事件發生之後,就會回撥註冊的介面。Reactor必要的幾個元件：Event Multiplexer事件分發器:即一些I/O複用機制select、po

資料結構與演算法之美專欄學習筆記-二叉樹基礎(上）

樹 節點的定義 樹中的元素稱之為節點 高度的定義 節點的高度：節點到葉子節點的最長路徑 樹的高度：跟節點的高度 深度的定義 根節點到這個節點所經歷的邊的個數 層的定義 節點的深度+1   二叉樹 滿二叉樹 除了葉子結點外每個節點都有左右兩個子節點 完全二叉樹 葉子結

資料結構與演算法之美專欄學習筆記-二叉樹基礎(下）

二叉查詢樹 Binary Search Tree  二叉查詢樹的定義 二叉查詢樹又稱二叉搜尋樹。其要求在二叉樹中的任意一個節點，其左子樹中的每個節點的值，都要小於這個節點的值，而右子樹的節點的值都大於這個節點的值。 二叉查詢樹的查詢操作 二叉樹類、節點類以及查詢方法的程式碼實現

數據結構與算法之美專欄學習筆記-二叉樹基礎(下）

binary 特性 child 數據大小 del delet 動態擴容 eve 怎麽 二叉查找樹 Binary Search Tree 二叉查找樹的定義 二叉查找樹又稱二叉搜索樹。其要求在二叉樹中的任意一個節點，其左子樹中的每個節點的值，都要小於這個節點的值，而右子樹的

《資料結構與演算法分析java語言描述》學習筆記二

一、抽象資料型別  抽象資料型別（abstract data type,ADT）是帶有一組操作的一些物件的集合。 二、ArrayList 的實現 自定義ArrayList 實現程式碼如下： package com.caitang.mjq; import java.uti

【opencv學習筆記二】opencv3.4.0元件結構說明

在學習opencv使用之前我們先來看一下opencv有哪些元件結構。至於OpenCV元件結構的研究方法，我們不妨管中窺豹，通過opencv安裝路徑下include目錄裡面標頭檔案的分類存放，來一窺OpenCV這些年迅猛發展起來的龐雜元件架構。我們進入到D:\ProgramFi

muduo學習筆記（二）

TCP分包 對於短連線，只要傳送方主動關閉連線，就表示一條訊息傳送完畢。 長連線有四種方法： 訊息長度固定； 使用特殊的字元或者字串作為訊息邊界，如http協議的headers以"\r\n"為欄位的分隔符 每條訊息的頭部加一個長度欄位 利用訊息本身的格式分包 mu

資料結構學習筆記——二叉樹之已知兩序排列還原二叉樹

對於非線性資料結構二叉樹，通過人為規定的三種遍歷順序將其轉化為線性結構存入計算機中。 三種遍歷順序即是三種轉換方式： 先序：先訪問當前節點，再訪問左子樹，後訪問右子樹。 中序：先訪問左子樹，再訪問當前節點，後訪問右子樹。 後序：先訪問左子樹，再訪問右子樹，後訪問當前節

TP5學習基礎二：目錄結構、URL路由、數據操作

更改 數據 valid nbsp 命名 別名 ret tag ati 一.安裝1.使用git或者composer(composer update)進行實時更新，區別在於git不會清空核心框架目錄而composer會清空。2.使用官網打包好的TP壓縮包(解壓即可用)->

SpringMVC學習筆記二：常用註解

title c學習 請求 pin 學習 lin att 詳解 stp 轉載請註明原文地址：http://www.cnblogs.com/ygj0930/p/6831976.html 一、用於定義類的註解：@Controller @Controller 用於標記在一個類上，

Scala學習筆記(二)(for循環相關)

spa nts multiple bool val turn 優化 n) 例子 Scala裏if...else語句 if語句不管是在哪種語言裏是使用最多的語句了. scala的if語句與java如出一轍. 舉個栗子就不再贅述: 一個 if 語句的語法： if(Boolean

ES6學習筆記(二)——字符串擴展

兩個 -m 開始 部分 學習筆記 erro xxx ocs 個數 相信很多人也和我一樣，不喜歡這樣循規蹈矩的逐條去學習語法，很枯燥乏味。主要是這樣學完一遍之後，沒過一段時間就忘到九霄雲外了。不如實際用到的時候研究它記得牢靠，所以我就整理成筆記，加深記憶的同時便於復習查看。

MySQL學習筆記(二）

發送 int 賦值 數據庫基礎知識 font 字符 需要 con spa -- 回顧 數據庫基礎知識： 關系型數據庫（磁盤），建立在關系模型上的數據庫，數據結構（二維表），浪費空間。 操作數據的指令集合：SQL(DDL,DML[DQL]和DCL) 完整性約束：表內和表之間（

Android第一行代碼學習筆記二---在活動中使用Toast

一行代碼 就會 onclick log sta contex instance and undle Toast：是Android系統提供的一種非常好的提醒方式，在程序中可以使用它將一些短小的信息通知給用戶，這些信息會在一段時間後自動消失，並且不會占用任何屏幕空間。 首先需要

framework7學習筆記二：基礎知識

部分 cnblogs query 基礎 logs code 自己 $$ pan 一：DOM7 framework7有自己的 DOM7 - 一個集成了大部分常用DOM操作的高性能庫。它的用法和jQuery幾乎是一樣的，包括大部分常用的方法和jquery風格的鏈式調用。 在開發

node.js學習筆記二之版本問題

nodejs targe tle 下一個 .cn blank 網站 mage 功能 一、版本說明 進入node.js官網https://nodejs.org/en/download/ 點擊上面的【All download options】進入到所有下載列表的地址 下載地

Java基礎學習筆記二十二 網絡編程

數據丟失 交互圖 主動 總結 交互 servers -- 處理 關閉 絡通信協議 通過計算機網絡可以使多臺計算機實現連接，位於同一個網絡中的計算機在進行連接和通信時需要遵守一定的規則，這就好比在道路中行駛的汽車一定要遵守交通規則一樣。在計算機網絡中，這些連接和通信的規則被

Java基礎學習筆記二十三 Java核心語法之反射

負責 目錄 boolean tostring 筆記 str 編譯 三種 進制 類加載器 類的加載 當程序要使用某個類時，如果該類還未被加載到內存中，則系統會通過加載，鏈接，初始化三步來實現對這個類進行初始化。 加載就是指將class文件讀入內存，並為之創建一個Clas

Java基礎學習筆記二十四 MySQL安裝圖解

password data 默認 count 重新 doc documents tran xp系統 、MYSQL的安裝 １、打開下載的mysql安裝文件mysql-5.5.27-win32.zip，雙擊解壓縮，運行“setup.exe”。