ACE学习八Reactor模式

这个模式为同步读+多线程处理的一个模型,在Windows下面默认的实现是ACE_WFMO_Reactor,他内部使用WaitForMuiltiObject来等待OVERLAPPED当中的event句柄,LINUX下使用ACE_Select_Reacotr实现,内部使用select函数来分配操作。

http://dl.vmall.com/c0bda5pwb4

Demo1

#include "ace/Auto_Ptr.h"
#include "ace/Log_Msg.h"
#include "ace/INET_Addr.h"
#include "ace/SOCK_Acceptor.h"
#include "ace/Reactor.h"
#include "ace/Message_Block.h"
#include "ace/Message_Queue.h"
#include "ace/SOCK_Stream.h"
#include "ace/Synch_Traits.h"
#include "ace/Synch.h"
#include "ace/OS_NS_sys_time.h"
#include "ace/os_include/os_netdb.h"


class ClientAcceptor : public ACE_Event_Handler
{
public:
	virtual ~ClientAcceptor();

	int open(const ACE_INET_Addr &listenAddr);

	// Get this handler's I/O handle.
	virtual ACE_HANDLE get_handle(void) const
	{
		return this->m_acceptor.get_handle();
	}

	// Called when a connection is ready to accept.
	virtual int handle_input(ACE_HANDLE fd = ACE_INVALID_HANDLE);

	// Called when this handler is removed from the ACE_Reactor.
	virtual int handle_close(ACE_HANDLE handle,ACE_Reactor_Mask closeMask);
protected:
	ACE_SOCK_Acceptor m_acceptor;
};

class ClientService : public ACE_Event_Handler
{
public:
	int open(void);
	ACE_SOCK_Stream &peer(void)
	{
		return this->m_sock;
	}

	// Get this handler's I/O handle.
	virtual ACE_HANDLE get_handle(void) const
	{
		return this->m_sock.get_handle();
	}

	// Called when input is available from the client
	virtual int handle_input(ACE_HANDLE fd = ACE_INVALID_HANDLE);

	// Called when output is possible.
	virtual int handle_output(ACE_HANDLE fd = ACE_INVALID_HANDLE);

	// Called when this handler is removed from the ACE_Reactor.
	virtual int handle_close(ACE_HANDLE handle,ACE_Reactor_Mask close_mask);

protected:
	ACE_SOCK_Stream m_sock;
	ACE_Message_Queue<ACE_NULL_SYNCH> m_output_queue;


};


int ClientAcceptor::open(const ACE_INET_Addr &listenAddr)
{
	if (this->m_acceptor.open(listenAddr,1) == -1)
	{
		ACE_ERROR_RETURN((LM_ERROR,ACE_TEXT("%p\n"),ACE_TEXT("acceptor.open")),-1);
	}
	/**
   * Register handler for I/O events.
   *
   * A handler can be associated with multiple handles. A handle
   * cannot be associated with multiple handlers.
   *
   * The handle will come from ACE_Event_Handler::get_handle().
   *
   * Reactor will call ACE_Event_Handler::add_reference() for a new
   * handler/handle pair.
   *
   * If this handler/handle pair has already been registered,any new
   * masks specified will be added. In this case,* ACE_Event_Handler::add_reference() will not be called.
   *
   * If the registered handler is currently suspended,it will remain
   * suspended.  When the handler is resumed,it will have the
   * existing masks plus any masks added through this call. Handlers
   * do not have partial suspensions.
   */

	// Get the event demultiplexors.
    // virtual ACE_Reactor *reactor (void) const;
	
	return this->reactor()->register_handler(this,ACE_Event_Handler::ACCEPT_MASK);
}

int ClientAcceptor::handle_input(ACE_HANDLE)
{
	// 为每次连接都使用单独的服务器处理器对象
	ClientService *client;
	ACE_NEW_RETURN (client,ClientService,-1);
	auto_ptr<ClientService> p(client);

	// client的peer方法返回的是一个ACE_SOCK_Stream对象
	if (this->m_acceptor.accept(client->peer()) == -1)
	{
		ACE_ERROR_RETURN((LM_ERROR,ACE_TEXT("(%P|%t) %p\n"),ACE_TEXT("Failed to accept"),ACE_TEXT("client connection")),-1);
	}

	p.release();
	client->reactor(this->reactor());

	// open方法会向反应器登记新得ClientService实例
	if (client->open() == -1)
	{
		client->handle_close(ACE_INVALID_HANDLE,0);
	}
	return 0;
}

int ClientAcceptor::handle_close(ACE_HANDLE handle,ACE_Reactor_Mask closeMask)
{
	if (this->m_acceptor.get_handle() != ACE_INVALID_HANDLE)
	{
		ACE_Reactor_Mask m = ACE_Event_Handler::ACCEPT_MASK | ACE_Event_Handler::DONT_CALL;
		this->reactor()->remove_handler(this,m);
		this->m_acceptor.close();
	}
	return 0;
}

ClientAcceptor::~ClientAcceptor()
{
	this->handle_close(ACE_INVALID_HANDLE,0);
}

//--------------------------------------------------
int ClientService::open(void)
{
	ACE_TCHAR peer_name[128];
	ACE_INET_Addr peer_addr;
	if (this->m_sock.get_remote_addr(peer_addr) == 0 &&
		peer_addr.addr_to_string(peer_name,128) == 0)
	{
		ACE_DEBUG((LM_DEBUG,ACE_TEXT("(%P|%t) Connection from %s\n"),peer_name));
	}

	return this->reactor()->register_handler(this,ACE_Event_Handler::READ_MASK);
}

int ClientService::handle_input(ACE_HANDLE)
{
	const size_t INPUT_SIZE = 4096;
	char buffer[INPUT_SIZE];
	ssize_t recv_cnt,send_cnt;

	if (recv_cnt = (this->m_sock.recv(buffer,sizeof(buffer))) <= 0)
	{
		ACE_DEBUG((LM_DEBUG,ACE_TEXT("(%P|%t) Conection closed\n")));
		return -1;
	}

	send_cnt = this->m_sock.send(buffer,ACE_static_cast(size_t,recv_cnt));
	if (send_cnt == recv_cnt)
	{
		return 0;
	}

	// EWOULBLOCK无法现在发送数据
	if (send_cnt == -1 && ACE_OS::last_error() != EWOULDBLOCK)
	{
		ACE_ERROR_RETURN((LM_ERROR,ACE_TEXT("send")),0);
	}

	if (send_cnt == -1)
	{
		send_cnt = 0;
	}

	ACE_Message_Block *mb;
	size_t remaining = ACE_static_cast(size_t,(recv_cnt - send_cnt));
	ACE_NEW_RETURN(mb,ACE_Message_Block(&buffer[send_cnt],remaining),-1);
	int output_off = this->m_output_queue.is_empty();
	ACE_Time_Value nowait(ACE_OS::gettimeofday());
	if (this->m_output_queue.enqueue_tail(mb,&nowait) == -1)
	{
		ACE_ERROR((LM_ERROR,ACE_TEXT("(%P|%t) %p; discarding data\n"),ACE_TEXT("enqueue failed")));
		mb->release();
		return 0;
	}

	if (output_off)
	{
		return this->reactor()->register_handler(this,ACE_Event_Handler::WRITE_MASK);
	}
	return 0;
}

int ClientService::handle_output(ACE_HANDLE)
{
	ACE_Message_Block *mb;
	ACE_Time_Value nowait(ACE_OS::gettimeofday());
	while (0 == this->m_output_queue.dequeue_head(mb,&nowait))
	{
		ssize_t send_cnt = this->m_sock.send(mb->rd_ptr(),mb->length());
		if (send_cnt == -1)
		{
			ACE_ERROR((LM_ERROR,ACE_TEXT("send")));
		}
		else
		{
			mb->rd_ptr(ACE_static_cast(size_t,send_cnt));
		}
		if(mb->length() > 0)
		{
			this->m_output_queue.enqueue_head(mb);
			break;
		}
		mb->release();

	}
	return (this->m_output_queue.is_empty()) ? -1:0;
}

int ClientService::handle_close(ACE_HANDLE,ACE_Reactor_Mask mask)
{
	if(mask == ACE_Event_Handler::WRITE_MASK)
		return 0;
	mask = ACE_Event_Handler::ALL_EVENTS_MASK |
		ACE_Event_Handler::DONT_CALL;
	this->reactor()->remove_handler(this,mask);
	this->m_sock.close();
	this->m_output_queue.flush();
	delete this;
	return 0;
}

int ACE_TMAIN(int,ACE_TCHAR *[])
{
	ACE_INET_Addr port_to_listen("5000");
	ClientAcceptor acceptor;

	// ACE_Event_Handler含有一个ACE_Reactor指针,用于方便的引用正在使用的反应器
	// 该反应器实例会在第一次用到的时候创建,在程序结束自动关闭,单例模式+智能指针?

	// Set the event demultiplexors.
    // virtual void reactor (ACE_Reactor *reactor);
	acceptor.reactor(ACE_Reactor::instance());

	if (acceptor.open(port_to_listen) == -1)
	{
		return 1;
	}

	ACE_Reactor::instance()->run_reactor_event_loop();
	return 0;
}


Demo2

// $Id: HAStatus.cpp 91626 2010-09-07 10:59:20Z johnnyw $

#include "ace/OS_NS_sys_time.h"
#include "ace/os_include/os_netdb.h"

// Listing 1 code/ch07
#include "ace/Auto_Ptr.h"
#include "ace/Log_Msg.h"
#include "ace/INET_Addr.h"
#include "ace/SOCK_Acceptor.h"
#include "ace/Reactor.h"

class ClientAcceptor : public ACE_Event_Handler
{
public:
  virtual ~ClientAcceptor ();

  //FUZZ: disable check_for_lack_ACE_OS
  int open (const ACE_INET_Addr &listen_addr);
  //FUZZ: enable check_for_lack_ACE_OS

  // Get this handler's I/O handle.
  virtual ACE_HANDLE get_handle (void) const
    { return this->acceptor_.get_handle (); }

  // Called when a connection is ready to accept.
  virtual int handle_input (ACE_HANDLE fd = ACE_INVALID_HANDLE);

  // Called when this handler is removed from the ACE_Reactor.
  virtual int handle_close (ACE_HANDLE handle,ACE_Reactor_Mask close_mask);

protected:
  ACE_SOCK_Acceptor acceptor_;
};
// Listing 1

// Listing 6 code/ch07
#include "ace/Message_Block.h"
#include "ace/Message_Queue.h"
#include "ace/SOCK_Stream.h"
#include "ace/Synch.h"

class ClientService : public ACE_Event_Handler
{
public:
  ACE_SOCK_Stream &peer (void) { return this->sock_; }

  //FUZZ: disable check_for_lack_ACE_OS
  int open (void);
  //FUZZ: enable check_for_lack_ACE_OS

  // Get this handler's I/O handle.
  virtual ACE_HANDLE get_handle (void) const
    { return this->sock_.get_handle (); }

  // Called when input is available from the client.
  virtual int handle_input (ACE_HANDLE fd = ACE_INVALID_HANDLE);

  // Called when output is possible.
  virtual int handle_output (ACE_HANDLE fd = ACE_INVALID_HANDLE);

  // Called when this handler is removed from the ACE_Reactor.
  virtual int handle_close (ACE_HANDLE handle,ACE_Reactor_Mask close_mask);

protected:
  ACE_SOCK_Stream sock_;
  ACE_Message_Queue<ACE_NULL_SYNCH> output_queue_;
};
// Listing 6

// Listing 5 code/ch07
ClientAcceptor::~ClientAcceptor ()
{
  this->handle_close (ACE_INVALID_HANDLE,0);
}
// Listing 5

// Listing 2 code/ch07
int
ClientAcceptor::open (const ACE_INET_Addr &listen_addr)
{
  if (this->acceptor_.open (listen_addr,1) == -1)
    ACE_ERROR_RETURN ((LM_ERROR,ACE_TEXT ("%p\n"),ACE_TEXT ("acceptor.open")),-1);
  return this->reactor ()->register_handler
    (this,ACE_Event_Handler::ACCEPT_MASK);
}
// Listing 2

// Listing 3 code/ch07
int
ClientAcceptor::handle_input (ACE_HANDLE)
{
  ClientService *client;
  ACE_NEW_RETURN (client,-1);
  auto_ptr<ClientService> p (client);

  if (this->acceptor_.accept (client->peer ()) == -1)
    ACE_ERROR_RETURN ((LM_ERROR,ACE_TEXT ("(%P|%t) %p\n"),ACE_TEXT ("Failed to accept ")
                       ACE_TEXT ("client connection")),-1);
  p.release ();
  client->reactor (this->reactor ());
  if (client->open () == -1)
    client->handle_close (ACE_INVALID_HANDLE,0);
  return 0;
}
// Listing 3

// Listing 4 code/ch07
int
ClientAcceptor::handle_close (ACE_HANDLE,ACE_Reactor_Mask)
{
  if (this->acceptor_.get_handle () != ACE_INVALID_HANDLE)
    {
      ACE_Reactor_Mask m = ACE_Event_Handler::ACCEPT_MASK |
                           ACE_Event_Handler::DONT_CALL;
      this->reactor ()->remove_handler (this,m);
      this->acceptor_.close ();
    }
  return 0;
}
// Listing 4

// Listing 7 code/ch07
int
ClientService::open (void)
{
  ACE_TCHAR peer_name[MAXHOSTNAMELEN];
  ACE_INET_Addr peer_addr;
  if (this->sock_.get_remote_addr (peer_addr) == 0 &&
      peer_addr.addr_to_string (peer_name,MAXHOSTNAMELEN) == 0)
    ACE_DEBUG ((LM_DEBUG,ACE_TEXT ("(%P|%t) Connection from %s\n"),peer_name));
  return this->reactor ()->register_handler
    (this,ACE_Event_Handler::READ_MASK);
}
// Listing 7

// Listing 8 code/ch07
int
ClientService::handle_input (ACE_HANDLE)
{
  const size_t INPUT_SIZE = 4096;
  char buffer[INPUT_SIZE];
  ssize_t recv_cnt,send_cnt;

  if ((recv_cnt = this->sock_.recv (buffer,sizeof(buffer))) <= 0)
    {
      ACE_DEBUG ((LM_DEBUG,ACE_TEXT ("(%P|%t) Connection closed\n")));
      return -1;
    }

  send_cnt =
    this->sock_.send (buffer,static_cast<size_t> (recv_cnt));
  if (send_cnt == recv_cnt)
    return 0;
  if (send_cnt == -1 && ACE_OS::last_error () != EWOULDBLOCK)
    ACE_ERROR_RETURN ((LM_ERROR,ACE_TEXT ("send")),0);
  if (send_cnt == -1)
    send_cnt = 0;
  ACE_Message_Block *mb = 0;
  size_t remaining =
    static_cast<size_t> ((recv_cnt - send_cnt));
  ACE_NEW_RETURN (mb,ACE_Message_Block (remaining),-1);
  mb->copy (&buffer[send_cnt],remaining);
  int output_off = this->output_queue_.is_empty ();
  ACE_Time_Value nowait (ACE_OS::gettimeofday ());
  if (this->output_queue_.enqueue_tail (mb,&nowait) == -1)
    {
      ACE_ERROR ((LM_ERROR,ACE_TEXT ("(%P|%t) %p; discarding data\n"),ACE_TEXT ("enqueue failed")));
      mb->release ();
      return 0;
    }
  if (output_off)
    return this->reactor ()->register_handler
      (this,ACE_Event_Handler::WRITE_MASK);
  return 0;
}
// Listing 8

// Listing 9 code/ch07
int
ClientService::handle_output (ACE_HANDLE)
{
  ACE_Message_Block *mb = 0;
  ACE_Time_Value nowait (ACE_OS::gettimeofday ());
  while (0 <= this->output_queue_.dequeue_head
                                    (mb,&nowait))
    {
      ssize_t send_cnt =
        this->sock_.send (mb->rd_ptr (),mb->length ());
      if (send_cnt == -1)
        ACE_ERROR ((LM_ERROR,ACE_TEXT ("send")));
      else
        mb->rd_ptr (static_cast<size_t> (send_cnt));
      if (mb->length () > 0)
        {
          this->output_queue_.enqueue_head (mb);
          break;
        }
      mb->release ();
    }
  return (this->output_queue_.is_empty ()) ? -1 : 0;
}
// Listing 9

// Listing 10 code/ch07
int
ClientService::handle_close (ACE_HANDLE,ACE_Reactor_Mask mask)
{
  if (mask == ACE_Event_Handler::WRITE_MASK)
    return 0;
  mask = ACE_Event_Handler::ALL_EVENTS_MASK |
         ACE_Event_Handler::DONT_CALL;
  this->reactor ()->remove_handler (this,mask);
  this->sock_.close ();
  this->output_queue_.flush ();
  delete this;
  return 0;
}
// Listing 10

// Listing 12 code/ch07
class LoopStopper : public ACE_Event_Handler
{
public:
  LoopStopper (int signum = SIGINT);

  // Called when object is signaled by OS.
  virtual int handle_signal (int signum,siginfo_t * = 0,ucontext_t * = 0);
};

LoopStopper::LoopStopper (int signum)
{
  ACE_Reactor::instance ()->register_handler (signum,this);
}

int
LoopStopper::handle_signal (int,siginfo_t *,ucontext_t *)
{
  ACE_Reactor::instance ()->end_reactor_event_loop ();
  return 0;
}
// Listing 12

// Listing 13 code/ch07
#include "ace/Signal.h"

class LogSwitcher : public ACE_Event_Handler
{
public:
  LogSwitcher (int on_sig,int off_sig);

  // Called when object is signaled by OS.
  virtual int handle_signal (int signum,ucontext_t * = 0);

  // Called when an exceptional event occurs.
  virtual int handle_exception (ACE_HANDLE fd = ACE_INVALID_HANDLE);

private:
  LogSwitcher () {}

  int on_sig_;       // Signal to turn logging on
  int off_sig_;      // Signal to turn logging off
  int on_off_;       // 1 == turn on,0 == turn off
};

LogSwitcher::LogSwitcher (int on_sig,int off_sig)
  : on_sig_ (on_sig),off_sig_ (off_sig)
{
  ACE_Sig_Set sigs;
  sigs.sig_add (on_sig);
  sigs.sig_add (off_sig);
  ACE_Reactor::instance ()->register_handler (sigs,this);
}
// Listing 13

// Listing 14 code/ch07
int
LogSwitcher::handle_signal (int signum,ucontext_t *)
{
  if (signum == this->on_sig_ || signum == this->off_sig_)
    {
      this->on_off_ = signum == this->on_sig_;
      ACE_Reactor::instance ()->notify (this);
    }
  return 0;
}
// Listing 14

// Listing 15 code/ch07
int
LogSwitcher::handle_exception (ACE_HANDLE)
{
  if (this->on_off_)
    ACE_LOG_MSG->clr_flags (ACE_Log_Msg::SILENT);
  else
    ACE_LOG_MSG->set_flags (ACE_Log_Msg::SILENT);
  return 0;
}
// Listing 15

// Listing 11 code/ch07
int ACE_TMAIN (int,ACE_TCHAR *[])
{
  ACE_INET_Addr port_to_listen ("HAStatus");
  ClientAcceptor acceptor;
  acceptor.reactor (ACE_Reactor::instance ());
  if (acceptor.open (port_to_listen) == -1)
    return 1;

  ACE_Reactor::instance ()->run_reactor_event_loop ();

  return (0);
}
// Listing 11


http://wenku.baidu.com/view/866966a1284ac850ad0242ab.html
http://daimojingdeyu.iteye.com/blog/828696
http://book.2cto.com/201208/1893.html
http://www.jb51.cc/article/p-nlrfvpff-yn.html
http://developer.51cto.com/art/201208/354591.htm

acceptor-connector

http://shanghai.kankanews.com/mobile/2013-05-20/1546541.shtml
http://auto.people.com.cn/n/2013/0317/c1005-20815213.html
http://scitech.people.com.cn/n/2013/0310/c1007-20736417.html

版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。

相关推荐


react 中的高阶组件主要是对于 hooks 之前的类组件来说的,如果组件之中有复用的代码,需要重新创建一个父类,父类中存储公共代码,返回子类,同时把公用属性...
我们上一节了解了组件的更新机制,但是只是停留在表层上,例如我们的 setState 函数式同步执行的,我们的事件处理直接绑定在了 dom 元素上,这些都跟 re...
我们上一节了解了 react 的虚拟 dom 的格式,如何把虚拟 dom 转为真实 dom 进行挂载。其实函数是组件和类组件也是在这个基础上包裹了一层,一个是调...
react 本身提供了克隆组件的方法,但是平时开发中可能很少使用,可能是不了解。我公司的项目就没有使用,但是在很多三方库中都有使用。本小节我们来学习下如果使用该...
mobx 是一个简单可扩展的状态管理库,中文官网链接。小编在接触 react 就一直使用 mobx 库,上手简单不复杂。
我们在平常的开发中不可避免的会有很多列表渲染逻辑,在 pc 端可以使用分页进行渲染数限制,在移动端可以使用下拉加载更多。但是对于大量的列表渲染,特别像有实时数据...
本小节开始前,我们先答复下一个同学的问题。上一小节发布后,有小伙伴后台来信问到:‘小编你只讲了类组件中怎么使用 ref,那在函数式组件中怎么使用呢?’。确实我们...
上一小节我们了解了固定高度的滚动列表实现,因为是固定高度所以容器总高度和每个元素的 size、offset 很容易得到,这种场景也适合我们常见的大部分场景,例如...
上一小节我们处理了 setState 的批量更新机制,但是我们有两个遗漏点,一个是源码中的 setState 可以传入函数,同时 setState 可以传入第二...
我们知道 react 进行页面渲染或者刷新的时候,会从根节点到子节点全部执行一遍,即使子组件中没有状态的改变,也会执行。这就造成了性能不必要的浪费。之前我们了解...
在平时工作中的某些场景下,你可能想在整个组件树中传递数据,但却不想手动地通过 props 属性在每一层传递属性,contextAPI 应用而生。
楼主最近入职新单位了,恰好新单位使用的技术栈是 react,因为之前一直进行的是 vue2/vue3 和小程序开发,对于这些技术栈实现机制也有一些了解,最少面试...
我们上一节了了解了函数式组件和类组件的处理方式,本质就是处理基于 babel 处理后的 type 类型,最后还是要处理虚拟 dom。本小节我们学习下组件的更新机...
前面几节我们学习了解了 react 的渲染机制和生命周期,本节我们正式进入基本面试必考的核心地带 -- diff 算法,了解如何优化和复用 dom 操作的,还有...
我们在之前已经学习过 react 生命周期,但是在 16 版本中 will 类的生命周期进行了废除,虽然依然可以用,但是需要加上 UNSAFE 开头,表示是不安...
上一小节我们学习了 react 中类组件的优化方式,对于 hooks 为主流的函数式编程,react 也提供了优化方式 memo 方法,本小节我们来了解下它的用...
开源不易,感谢你的支持,❤ star me if you like concent ^_^
hel-micro,模块联邦sdk化,免构建、热更新、工具链无关的微模块方案 ,欢迎关注与了解
本文主题围绕concent的setup和react的五把钩子来展开,既然提到了setup就离不开composition api这个关键词,准确的说setup是由...
ReactsetState的执行是异步还是同步官方文档是这么说的setState()doesnotalwaysimmediatelyupdatethecomponent.Itmaybatchordefertheupdateuntillater.Thismakesreadingthis.staterightaftercallingsetState()apotentialpitfall.Instead,usecom