es6写 react 用
2014-07-01 19:42:45 iw1210 阅读数 4395

用ACE的Reactor模式实现网络通讯的例子,不罗嗦,直接上代码。

服务器代码:

#include <ace/Reactor.h>
#include <ace/SOCK_Connector.h> 
#include <ace/SOCK_Acceptor.h> 
#include <ace/Auto_Ptr.h>


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

    int open (void)
    {
        //注册读就绪回调函数
        return this->reactor ()->register_handler(this, ACE_Event_Handler::READ_MASK);
    }

    virtual ACE_HANDLE get_handle (void) const { return this->sock_.get_handle (); }

    virtual int handle_input (ACE_HANDLE fd )
    {
        int rev = peer().recv(buf,sizeof(buf));
        if(rev<=0)
            return -1;
		buf[rev] = '\0';
		printf("recv: %s",buf);
		strcpy(buf,"hello,Client\n");
        peer().send(buf,strlen(buf)); //向客户端发送信息。
        return 0;
    }

    // 释放相应资源
virtual int 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 ();
        delete this;    //socket出错时,将自动删除该客户端,释放相应资源
        return 0;
    }

protected:
    char buf[100];
    ACE_SOCK_Stream sock_;
};


class ClientAcceptor : public ACE_Event_Handler
{
public:
    virtual ~ClientAcceptor (){this->handle_close (ACE_INVALID_HANDLE, 0);}

    int open (const ACE_INET_Addr &listen_addr)
    {
        if (this->acceptor_.open (listen_addr, 1) == -1)
        {
            printf("open port fail\n");
            return -1;
        }
        //注册接受连接回调事件
        return this->reactor ()->register_handler(this, ACE_Event_Handler::ACCEPT_MASK);
    }

    virtual ACE_HANDLE get_handle (void) const
    { return this->acceptor_.get_handle (); }

    virtual int handle_input (ACE_HANDLE fd )
    {
        ClientService *client = new ClientService();
        auto_ptr<ClientService> p (client);

        if (this->acceptor_.accept (client->peer ()) == -1)
        {
            printf("accept client fail\n");
            return -1;
        }
        p.release ();
        client->reactor (this->reactor ());
        if (client->open () == -1)
            client->handle_close (ACE_INVALID_HANDLE, 0);
        return 0;
    }
    
    virtual int handle_close (ACE_HANDLE handle,
        ACE_Reactor_Mask close_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;
    }

protected:
    ACE_SOCK_Acceptor acceptor_;
};

int main(int argc, char *argv[]) 
{
    ACE_INET_Addr addr(3000,"127.0.0.1");
    ClientAcceptor server;
    server.reactor(ACE_Reactor::instance());
    server.open(addr);

    while(true)
    {
        ACE_Reactor::instance()->handle_events(); 
    }

    return 0; 
} 


客户端代码:

#include "ace/Reactor.h"
#include "ace/SOCK_Connector.h"

#include <string>
#include <iostream>
using namespace std;

class MyClient:public ACE_Event_Handler 
{
public:
    bool open()
    {
        ACE_SOCK_Connector connector;
        ACE_INET_Addr addr(3000,"127.0.0.1");
        ACE_Time_Value timeout(5,0);
        if(connector.connect(peer,addr,&timeout) != 0)
        {
            cout<<endl<<"connect fail.";
            return false;
        }
        ACE_Reactor::instance()->register_handler(this,ACE_Event_Handler::READ_MASK);
        cout<<endl<<"connected.";
		strcpy(buf, "hello,Server\n");
		peer.send(buf,strlen(buf));
        return true;
    }

    ACE_HANDLE get_handle(void) const
    {
        return peer.get_handle();
    }

    int handle_input (ACE_HANDLE fd)
    {
        int rev=0;
        ACE_Time_Value timeout(5,0);
        if((rev=peer.recv(buf,sizeof(buf),&timeout))>0)
        {
            buf[rev]='\0';
            cout<<endl<<"recv: "<<buf<<endl;
        }
        return 3;
    }

private:
    ACE_SOCK_Stream peer;
    char buf[100];
};

int main(int argc, char *argv[]) 
{
    MyClient client;
    client.open();

    while(true)
    {
        ACE_Reactor::instance()->handle_events(); 
    }

    return 0; 
}


分别编译运行(先运行服务端,后运行客户端)下面是执行结果。

服务器:

recv: hello,Server

客户端:

connected.
recv: hello,Client

 

参考:

http://www.cnblogs.com/TianFang/archive/2006/12/13/591332.html

http://www.cnblogs.com/TianFang/archive/2006/12/18/595938.html

es6写 react 用 相关内容

2017-12-20 21:54:39 shixin_0125 阅读数 146

用ACE的Reactor模式实现网络通讯的例子,不罗嗦,直接上代码。

服务器代码:

  1. #include <ace/Reactor.h>  
  2. #include <ace/SOCK_Connector.h>   
  3. #include <ace/SOCK_Acceptor.h>   
  4. #include <ace/Auto_Ptr.h>  
  5.   
  6.   
  7. class ClientService : public ACE_Event_Handler  
  8. {  
  9. public:  
  10.     ACE_SOCK_Stream &peer (void) { return this->sock_; }  
  11.   
  12.     int open (void)  
  13.     {  
  14.         //注册读就绪回调函数  
  15.         return this->reactor ()->register_handler(this, ACE_Event_Handler::READ_MASK);  
  16.     }  
  17.   
  18.     virtual ACE_HANDLE get_handle (voidconst { return this->sock_.get_handle (); }  
  19.   
  20.     virtual int handle_input (ACE_HANDLE fd )  
  21.     {  
  22.         int rev = peer().recv(buf,sizeof(buf));  
  23.         if(rev<=0)  
  24.             return -1;  
  25.         buf[rev] = '\0';  
  26.         printf("recv: %s",buf);  
  27.         strcpy(buf,"hello,Client\n");  
  28.         peer().send(buf,strlen(buf)); //向客户端发送信息。  
  29.         return 0;  
  30.     }  
  31.   
  32.     // 释放相应资源  
  33. virtual int handle_close (ACE_HANDLE, ACE_Reactor_Mask mask)  
  34.     {  
  35.         if (mask == ACE_Event_Handler::WRITE_MASK)  
  36.             return 0;  
  37.         mask = ACE_Event_Handler::ALL_EVENTS_MASK |  
  38.             ACE_Event_Handler::DONT_CALL;  
  39.         this->reactor ()->remove_handler (this, mask);  
  40.         this->sock_.close ();  
  41.         delete this;    //socket出错时,将自动删除该客户端,释放相应资源  
  42.         return 0;  
  43.     }  
  44.   
  45. protected:  
  46.     char buf[100];  
  47.     ACE_SOCK_Stream sock_;  
  48. };  
  49.   
  50.   
  51. class ClientAcceptor : public ACE_Event_Handler  
  52. {  
  53. public:  
  54.     virtual ~ClientAcceptor (){this->handle_close (ACE_INVALID_HANDLE, 0);}  
  55.   
  56.     int open (const ACE_INET_Addr &listen_addr)  
  57.     {  
  58.         if (this->acceptor_.open (listen_addr, 1) == -1)  
  59.         {  
  60.             printf("open port fail\n");  
  61.             return -1;  
  62.         }  
  63.         //注册接受连接回调事件  
  64.         return this->reactor ()->register_handler(this, ACE_Event_Handler::ACCEPT_MASK);  
  65.     }  
  66.   
  67.     virtual ACE_HANDLE get_handle (voidconst  
  68.     { return this->acceptor_.get_handle (); }  
  69.   
  70.     virtual int handle_input (ACE_HANDLE fd )  
  71.     {  
  72.         ClientService *client = new ClientService();  
  73.         auto_ptr<ClientService> p (client);  
  74.   
  75.         if (this->acceptor_.accept (client->peer ()) == -1)  
  76.         {  
  77.             printf("accept client fail\n");  
  78.             return -1;  
  79.         }  
  80.         p.release ();  
  81.         client->reactor (this->reactor ());  
  82.         if (client->open () == -1)  
  83.             client->handle_close (ACE_INVALID_HANDLE, 0);  
  84.         return 0;  
  85.     }  
  86.       
  87.     virtual int handle_close (ACE_HANDLE handle,  
  88.         ACE_Reactor_Mask close_mask)  
  89.     {  
  90.         if (this->acceptor_.get_handle () != ACE_INVALID_HANDLE)  
  91.         {  
  92.             ACE_Reactor_Mask m = ACE_Event_Handler::ACCEPT_MASK |  
  93.                 ACE_Event_Handler::DONT_CALL;  
  94.             this->reactor ()->remove_handler (this, m);  
  95.             this->acceptor_.close ();  
  96.         }  
  97.         return 0;  
  98.     }  
  99.   
  100. protected:  
  101.     ACE_SOCK_Acceptor acceptor_;  
  102. };  
  103.   
  104. int main(int argc, char *argv[])   
  105. {  
  106.     ACE_INET_Addr addr(3000,"127.0.0.1");  
  107.     ClientAcceptor server;  
  108.     server.reactor(ACE_Reactor::instance());  
  109.     server.open(addr);  
  110.   
  111.     while(true)  
  112.     {  
  113.         ACE_Reactor::instance()->handle_events();   
  114.     }  
  115.   
  116.     return 0;   
  117. }   


客户端代码:

  1. #include "ace/Reactor.h"  
  2. #include "ace/SOCK_Connector.h"  
  3.   
  4. #include <string>  
  5. #include <iostream>  
  6. using namespace std;  
  7.   
  8. class MyClient:public ACE_Event_Handler   
  9. {  
  10. public:  
  11.     bool open()  
  12.     {  
  13.         ACE_SOCK_Connector connector;  
  14.         ACE_INET_Addr addr(3000,"127.0.0.1");  
  15.         ACE_Time_Value timeout(5,0);  
  16.         if(connector.connect(peer,addr,&timeout) != 0)  
  17.         {  
  18.             cout<<endl<<"connect fail.";  
  19.             return false;  
  20.         }  
  21.         ACE_Reactor::instance()->register_handler(this,ACE_Event_Handler::READ_MASK);  
  22.         cout<<endl<<"connected.";  
  23.         strcpy(buf, "hello,Server\n");  
  24.         peer.send(buf,strlen(buf));  
  25.         return true;  
  26.     }  
  27.   
  28.     ACE_HANDLE get_handle(voidconst  
  29.     {  
  30.         return peer.get_handle();  
  31.     }  
  32.   
  33.     int handle_input (ACE_HANDLE fd)  
  34.     {  
  35.         int rev=0;  
  36.         ACE_Time_Value timeout(5,0);  
  37.         if((rev=peer.recv(buf,sizeof(buf),&timeout))>0)  
  38.         {  
  39.             buf[rev]='\0';  
  40.             cout<<endl<<"recv: "<<buf<<endl;  
  41.         }  
  42.         return 3;  
  43.     }  
  44.   
  45. private:  
  46.     ACE_SOCK_Stream peer;  
  47.     char buf[100];  
  48. };  
  49.   
  50. int main(int argc, char *argv[])   
  51. {  
  52.     MyClient client;  
  53.     client.open();  
  54.   
  55.     while(true)  
  56.     {  
  57.         ACE_Reactor::instance()->handle_events();   
  58.     }  
  59.   
  60.     return 0;   
  61. }  


分别编译运行(先运行服务端,后运行客户端)下面是执行结果。

服务器:

recv: hello,Server

客户端:

connected.
recv: hello,Client

 

参考:

http://www.cnblogs.com/TianFang/archive/2006/12/13/591332.html

http://www.cnblogs.com/TianFang/archive/2006/12/18/595938.html

es6写 react 用 相关内容

2006-11-07 15:07:00 haohao_h 阅读数 1148

#include <ace/OS_main.h>
#include <ace/ACE.h>
#include <ace/Svc_Handler.h>
#include <ace/Acceptor.h>
#include <ace/Log_Msg.h>
#include <ace/SOCK_Acceptor.h>
#include <ace/Thread.h>

class My_Svc_Handler:
    public ACE_Svc_Handler <ACE_SOCK_STREAM,ACE_NULL_SYNCH>
{
    public:
        int open(void*)
        {
            ACE_INET_Addr raddr;
            peer().get_remote_addr(raddr);//get the client addr
            reactor()->register_handler(this,ACE_Event_Handler::READ_MASK);   
            return 0;
        }

        int handle_input(ACE_HANDLE)
        {
            char buffer[64] = {0};
            int len = peer().recv(buffer,sizeof(buffer)-1);
            printf("msg:%s/n",buffer);
            if(len>0)
            {
                return 0;
            }
            else
            {
                return -1;//here to call handle_close() function to disconnect client's linker.
            }
        }

        int handle_close(ACE_HANDLE handle,ACE_Reactor_Mask close_mask)
        {
            ACE_INET_Addr raddr;
            peer().get_remote_addr(raddr);
            ACE_DEBUG ((LM_DEBUG,ACE_TEXT ("(%P|%t) close:%s %d/n"),raddr.get_host_addr(),raddr.get_port_number()));
            peer().close();
            return 0;
        }
};

#define PORT_NUM 10101

typedef ACE_Acceptor<My_Svc_Handler,ACE_SOCK_ACCEPTOR> MyAcceptor;
int main(int argc, char *argv[])
{
    ACE_INET_Addr addr(PORT_NUM);
    MyAcceptor server;

    if(server.open(addr)==-1)
    {
        ACE_DEBUG ((LM_DEBUG,"fail to bind/n"));
        return 1;
    }

    while(1){
        ACE_Reactor::instance()->handle_events();
    }
    server.close();
    return 0;

es6写 react 用 相关内容

2017-08-19 22:01:19 zhanyuanlin 阅读数 1717

1.前言

从Yarn RPC Server到Kafka Server,凡是有高并发需求的服务端,无一例外采用了基于了Reactor设计模式。在我的博客《Hadoop RPC Server基于Reactor模式和Java NIO 的架构和原理》中,分析了Yarn的基于Reactor设计模式和Java NIO实现的RPC Server的架构和设计,而Kafka的Server端网络层也同样使用了Reactor设计模式。Reactor模式有以下显著特定:
1. 通过事件派发、分离IO操作,来提高系统的并发特性
2. 提供粗粒度的并发控制,使用单线程实现,避免了复杂的并发处理

下文中,我们就从Kafka Server端网络层代码入手,分析和讲解Kafka基于Ractor模式的具体网络层实现以及网络层和具体的业务层的衔接逻辑。在我的另外一篇博客《基于Java NIO的Kafka底层网络层源码和架构》中,已经详细讲解了Kafka使用Java NIO实现的底层网络架构,本文讲解Reactor设计模式,将忽略具体的网络层,而将注意力放在NIO的上层,即网络事件的派发、请求和响应的处理上面。

同样,事先说明,Kafka自己对NIO的Selector进行了封装,放在了org.apache.kafka.common.network.Selector,为了与java原生的java.nio.channels.Selector区分,我参照《Apache Kafka源码剖析》一书的叫法,将org.apache.kafka.common.network.Selector叫做KSelector。Kafka的客户端与服务器端的通信以及服务器之间的通信,底层都是用KSelector进行。

为了便于在代码层理解,我画了整个KafkaServer从网络层消息到业务处理,以及相反,业务处理结果通过网络层返回的基于Reactor模式的处理流如下:

从网络层接收请求到交付给业务处理流程示意图:
这里写图片描述

从业务处理层返回结果,到将结果通过网络层返回给客户端的流程图:
这里写图片描述


2. SocketServer初始化和启动

Kafka的网络层入口类是SocketServer。
我们知道,kafka.Kafka是Kafka Broker的入口类,kafka.Kafka.main()是Kafka Server的main()方法,即Kafka Broker的启动入口。我们跟踪代码,即沿着方法调用栈kafka.Kafka.main() -> KafkaServerStartable() -> KafkaServer().startup可以从main()方法入口一直跟踪到SocketServer即网络层对象的创建,这意味着Kafka Server启动的时候会初始化并启动SocketServer,即我们所熟知的打开9092端口,这样,SocketServer启动完毕以后,就可以通过9092端口监听、解析、处理所有请求。

SocketServer的核心任务,是完成9092端口的绑定,监听并处理客户端或者其它Kafka Broker的请求,将请求分派,然后将相应返回给发起请求的对应客户端或者其它的Kafka Broker。核心角色,是同基本Reactor模式一致的负责接收请求的Acceptor角色、负责具体请求管理的Processor角色、负责请求和响应队列的RequestChannel角色以及负责管理、限制整个网络负载的ConnectionQuotas角色。

 private val endpoints = config.listeners
  private val numProcessorThreads = config.numNetworkThreads
  private val maxQueuedRequests = config.queuedMaxRequests
  private val totalProcessorThreads = numProcessorThreads * endpoints.size
  private val maxConnectionsPerIp = config.maxConnectionsPerIp
  private val maxConnectionsPerIpOverrides = config.maxConnectionsPerIpOverrides
  //创建RequestChannel,有totalProcessorThreads个responseQueue队列,
  val requestChannel = new RequestChannel(totalProcessorThreads, maxQueuedRequests)
  //所有的processers,长度为totalProcessorThreads
  private val processors = new Array[Processor](totalProcessorThreads)

def startup() {
    this.synchronized {

      connectionQuotas = new ConnectionQuotas(maxConnectionsPerIp, maxConnectionsPerIpOverrides)

      val sendBufferSize = config.socketSendBufferBytes
      val recvBufferSize = config.socketReceiveBufferBytes
      val brokerId = config.brokerId

      var processorBeginIndex = 0
      endpoints.values.foreach { endpoint =>//遍历endPoint集合,对于每一个endpoint,创建一个acceptor和多个processor
        val protocol = endpoint.protocolType
        val processorEndIndex = processorBeginIndex + numProcessorThreads  //按照序号,创建process

        for (i <- processorBeginIndex until processorEndIndex)
          processors(i) = newProcessor(i, connectionQuotas, protocol)

        val acceptor = new Acceptor(endpoint, sendBufferSize, recvBufferSize, brokerId,
          processors.slice(processorBeginIndex, processorEndIndex), connectionQuotas)
        acceptors.put(endpoint, acceptor)//保存endr的point和accepto对应关系
        //起一个名字叫做"kafka-socket-acceptor-%s-%d".format(protocol.toString, endpoint.port)的acceptor线程,非deamon
        Utils.newThread("kafka-socket-acceptor-%s-%d".format(protocol.toString, endpoint.port), acceptor, false).start()
        acceptor.awaitStartup()//一直等到acceptor的run()方法的第一条语句开始执行,才证明已经启动

        processorBeginIndex = processorEndIndex
      }
    }

SocketServer的启动方法中,会遍历本机的所有EndPoint(一个EndPoint一般对应一个网卡),为每一个EndPoint创建一个唯一独立的SocketServer.Acceptor对象,负责处理这个EndPoint上的所有请求。我们从代码private val endpoints = config.listeners可以看到EndPoint是通过server.properties的listeners中进行配置的,用户可以配置一个或者多个listener。

  • numProcessorThreads:通过num.network.threads进行配置,单个Acceptor所管理的Processor对象的数量;
  • maxQueuedRequests:通过queued.max.requests进行配置,请求队列所允许的最大的未响应请求的数量,用来给ConnectionQuotas进行请求限额控制,避免Kafka Server产生过大的网络负载;
  • totalProcessorThreads:计算方式为numProcessorThreads * endpoints.size,即单台机器总的Processor的数量;
  • maxConnectionsPerIp:配置项为max.connections.per.ip,单个IP上的最大连接数,用来给ConnectionQuotas控制连接数;
  • maxConnectionsPerIpOverrides:针对每个IP进行的个性化限制,因为每个IP的最大连接数限额可能会差异化,因此可以通过此项覆盖maxConnectionsPerIp中的全局统一配置,这是在KAFKA-1512中增加的improvement;

同时,负责创建一个空的Processor对象的数组,用来放置所有的Processor线程对象,但是并未实际创建Processor对象。Processor是由Acceptor直接管理,因此也是由SocketServer.Acceptor负责创建的,这符合角色分层的原则。角色分层在Yarn的代码中也体现得非常好,这样做可以让一个复杂的系统的每个功能模块都变得清晰可控。每层角色只会负责自己还有自己直接管理的直接下层角色的初始化和启动,绝对不会去触碰不是自己所直接管理的角色。

根据Acceptor的数量,将这些线程对象平均分配给Acceptor。然后,一直等待Acceptor启动完成,SocketServer.startup()会一直阻塞等待启动完成才会退出代表SocketServer完成启动。阻塞方式我们在SocketServer.Acceptor的讲解中会进行分析。总之,下层多个服务全部启动完成,上层服务才算启动完成,这也是职责分层的原则。


3.SocketServer.Acceptor

每一个SocketServer.Acceptor负责唯一一个endpoint上的网络请求和响应的管理。正常情况下,我们会给的Kafka Server配置唯一的一个Endpoint。但是有些服务器具有多个网卡驱动,因此可以配置多个EndPoint。旧版本的Kafka使用host.name来标识集群中的主机身份,到了新版本的kakfa,则改为使用listenersadvertised.listeners 进行配置,这主要用于复杂网络环境下的Kafka配置,比如,我们的Kafka Server安装在亚马逊云上,三台服务器,每台服务器都有一个内网IP和外网IP,Server之间的通信走内网IP以节省网络流量,而外网用户的访问则只能使用公网IP,这时候我们就可以将内网IP的host配置在listeners中,而将外网IP配置在advertised.listeners 中。在下文中,我们默认一台机器只有一个网卡的基本情况,以避免不必要的麻烦。

private[kafka] class Acceptor(val endPoint: EndPoint,
                              val sendBufferSize: Int,
                              val recvBufferSize: Int,
                              brokerId: Int,
                              processors: Array[Processor],
                              connectionQuotas: ConnectionQuotas) extends AbstractServerThread(connectionQuotas) with KafkaMetricsGroup {

  private val nioSelector = NSelector.open()
  val serverChannel = openServerSocket(endPoint.host, endPoint.port)//创建一个ServerSocketChannel,监听endPoint.host, endPoint.port套接字

  //Acceptor被构造的时候就会启动所有的processor线程
  this.synchronized {
    //每个processor创建一个单独线程
    processors.foreach { processor =>
      Utils.newThread("kafka-network-thread-%d-%s-%d".format(brokerId, endPoint.protocolType.toString, processor.id), processor, false).start()
    }
  }

Acceptor的构造方法中,首先通过openServerSocket()打开自己负责的EndPoint的Socket,即打开端口并启动监听。
然后,Acceptor会负责构造自己管理的一个或者多个Processor对象。其实,每一个Processor都是一个独立线程,下面在介绍Processor的时候会说到。

Acceptor使用CountDownLatch对象来标记启动或者关闭是否完成:

Acceptor完成启动,会调用

  /**
   * Record that the thread startup is complete
   */
  protected def startupComplete() = {
    startupLatch.countDown() 
  }

方法将startupLatch置位为0,通知上层SocketServer自己已经完成启动。此时上层SocketServer正通过调用startupLatch.await一直阻塞等待当前acceptor完成初始化。当SocketServer管理的所有Acceptor均完成了启动,SocketServer完成启动。

  def run() {
    //向selector注册channel,可以接收ACCEPT事件,只有非阻塞的serverChannel才可以注册给Selector
    serverChannel.register(nioSelector, SelectionKey.OP_ACCEPT)
    startupComplete() //启动完成的标记,放在run()方法的第一行,说明当确认线程开始运行,则认为启动成功,启动过以后就是运行了
    try {
      var currentProcessor = 0
      while (isRunning) {///无限循环,持续等待OP_ACCEPT事件发生
        try {
          val ready = nioSelector.select(500)
          if (ready > 0) {//已经有对应的Accept事件发生
            val keys = nioSelector.selectedKeys()//取出发了了对应事件的事件的key,即有连接事件发生
            val iter = keys.iterator()
            while (iter.hasNext && isRunning) {
              try {
                val key = iter.next
                iter.remove()//NIO的通用做法,取出一个有相关事件发生的channel以后,必须remove掉对应的SelectionKey,防止下次重复取出
                if (key.isAcceptable)//
                  accept(key, processors(currentProcessor))//通过round-robin的方式取出一个acceptor进行处理
                else
                  throw new IllegalStateException("Unrecognized key state for acceptor thread.")

                //round-robin序号增1,下一个连接会取出下一个processor
                currentProcessor = (currentProcessor + 1) % processors.length
              } catch {
                case e: Throwable => error("Error while accepting connection", e)
              }
            }
          }
        }
        catch { //略  }
      }
    } finally { //略 }
  }

Acceptor线程的run()方法,是不断监听对应ServerChannel上的连接请求,如果有新的连接请求,就选择出一个Processor,用来处理这个请求,将这个新连接交付给Processor是在方法Acceptor.accept()中:

 def accept(key: SelectionKey, processor: Processor) {
    val serverSocketChannel = key.channel().asInstanceOf[ServerSocketChannel]//取出channel
    val socketChannel = serverSocketChannel.accept()//创建socketChannel,专门负责与这个客户端的连接
    try {
      //socketChannel参数设置
      processor.accept(socketChannel)//将SocketChannel交给process进行处理
    } catch {
      //异常处理
    }
  }

//Processor.accept():
 /**
   * Queue up a new connection for reading
   */
  def accept(socketChannel: SocketChannel) {
    newConnections.add(socketChannel)
    wakeup()
  }

Processor将这个新的SocketChannel加入到自己维护的ConcurrentLinkedQueue[SocketChannel] newConnections中。
下面介绍Processor线程.

4.Processor

Processor负责不断检查Acceptor是否有交付给自己新的连接,如果有,就负责这个channel上消息的读写操作,即:监听自己维护的所有channel上的读请求,解析并交付给对应的业务处理逻辑;监听新的写操作,把服务端的响应数据通过SocketChannel准确地发送给客户端。

注意,每一个Processor都维护了一个单独的KSelector对象,这个KSelector只负责这个Processor上所有channel的监听。这样最大程度上保证了不同Processor线程之间的完全并行和业务隔离,尽管,在异步IO情况下,一个Selector负责成百上千个socketChannel的状态监控也不会带来效率问题。

  override def run() {
    startupComplete()//表示初始化流程已经结束,通过这个CountDownLatch代表初始化已经结束,这个Processor已经开始正常运行了
    while (isRunning) {
      try {
        // setup any new connections that have been queued up
        configureNewConnections()//为已经接受的请求注册OR_READ事件
        // register any new responses for writing
        processNewResponses()//处理响应队列,这个响应队列是Handler线程处理以后的结果,会交付给RequestChannel.responseQueue.同时调用unmute,开始接受请求
        poll()  //调用KSelector.poll(),进行真正的数据读写
        processCompletedReceives()//调用mute,停止接受新的请求
        processCompletedSends()
        processDisconnected()
      } catch {
        //异常处理 略
    }

    debug("Closing selector - processor " + id)
    swallowError(closeAll())
    shutdownComplete()
  }

Processor.run()方法中,Processor主要进行以下几步操作:
1. 新连接的处理:通过configureNewConnections()来处理Acceptor交付给自己的新的连接请求,即:把在自己的Selector上注册这个SocketChannel,并监听SelectionKey.OP_READ操作:
2. 响应处理:Processor如何将响应交付给业务层进行处理的我在后面会讲到,在这里,Processor通过processNewResponses()方法,从RequestChannel.responseQueue中取出当前Processor的一个Response,通过对应的SocketChannel,发送给Client。注意,这里的”发送”,其实并没有真正发送,只是通过调用KafkaChannel.send()方法,将待发送数据交付给对应的Channel,真正的数据发送是调用KafkaChannel.write()方法,这是在下面第三步的poll()中进行的;关于RequestChannel的职责,下文中会详细讲解。
3. 响应处理完毕,开始进行真正的读写操作,即监控自己负责的所有SocketChannel,是否有新的读请求进来,如果有,就进行相应的处理,主要是对请求进行解析并交付给业务逻辑进行具体调用,如果有待写数据,就执行写操作,真正将数据发送给远程;如果有新的连接,则放到connected中,如果有断开的连接,则放入到disconnected中:

  • List<String> connected:已经建立连接的KafkaChannel的id的list
  • List<String> disconnected:出现异常的KafkaChannel的id的list
  • List<Send> completedSends:已经完成的发送请求,服务端可以通过查询completedSends,知道哪些请求已经发送出去
  • List<NetworkReceive> completedReceives:已经接收到的数据,服务端拿到这些已经接收到的数据以后,会将这些数据交付给具体数据的处理者进行处理

我们先来看Processor对于已经收到的请求,是如何交付给业务端进行处理的:


   * 将completedReceived中的对象进行封装,交付给requestQueue.completRequets
   */
  private def processCompletedReceives() {
    selector.completedReceives.asScala.foreach { receive =>//每一个receive是一个NetworkReceivedui'xiagn
      try {
        //receive.source代表了这个请求的发送者的身份,KSelector保存了channel另一端的身份和对应的SocketChannel之间的对应关系
        val channel = selector.channel(receive.source)
        val session = RequestChannel.Session(new KafkaPrincipal(KafkaPrincipal.USER_TYPE, channel.principal.getName),
          channel.socketAddress)
        val req = RequestChannel.Request(processor = id, connectionId = receive.source, session = session, buffer = receive.payload, startTimeMs = time.milliseconds, securityProtocol = protocol)
        requestChannel.sendRequest(req)//将请求通过RequestChannel.requestQueue交付给Handler
        selector.mute(receive.source)//不再接受Read请求,发送响应之前,不可以再接收任何请求
      } catch {
        //异常处理 略
      }
    }
  }

Processor.processCompletedReceives()通过遍历completedReceives,对于每一个已经完成接收的数据,对数据进行解析和封装,交付给RequestChannel,RequestChannel会交付给具体的业务处理层进行处理。
从上面的分析可以知道,Processor从RequetsChannel中获取业务层的处理结果然后将结果发送给客户端,同时,收到远程的某些请求,也是交付给RequetsChannel进行处理。可见,RequetsChannel负责了网络层和业务层的数据交付。下文中,我们一起来看RequestChannel的工作机制。


5. RequestChannel

RequestChannel负责消息从网络层转接到业务层,以及将业务层的处理结果交付给网络层进而返回给客户端。每一个SocketServer只有一个RequestChannel对象,在SocketServer中构造。

  //创建RequestChannel,有totalProcessorThreads个responseQueue队列,
  val requestChannel = new RequestChannel(totalProcessorThreads, maxQueuedRequests)

我们一起来看RequestChannel的构造方法:

class RequestChannel(val numProcessors: Int, val queueSize: Int) extends KafkaMetricsGroup {
  private var responseListeners: List[(Int) => Unit] = Nil
  //request存放了所有Processor接收到的远程请求,负责把requestQueue中的请求交付给具体业务逻辑进行处理
  private val requestQueue = new ArrayBlockingQueue[RequestChannel.Request](queueSize)
  //responseQueues存放了所有Processor的带出来的response,即每一个Processor都有一个response queue
  private val responseQueues = new Array[BlockingQueue[RequestChannel.Response]](numProcessors)
  for(i <- 0 until numProcessors) //初始化responseQueues
    responseQueues(i) = new LinkedBlockingQueue[RequestChannel.Response]()

  //一些metrics用来监控request和response的数量,代码略
  }

RequestChannel构造方法中初始化了requestQueue,用来存放网络层接收到的请求,这些请求即将交付给业务层进行处理。同时,初始化了responseQueues,为每一个Processor建立了一个response队列,用来存放这个Processor的一个或者多个Response,这些response即将交付给网络层返回给客户端。

6. KafkaRequestHandler线程和KafkaRequestHandlerPool线程池

在上面讲到Processor.processCompletedReceives()方法时,可以看到Processor调用了RequestChannel.sendRequest()方法将请求交付给了RequestChannel。而从RequestChannel取出请求并交付业务层处理的逻辑,是在独立线程中完成的,它是KafkaRequestHandler。并且,KafkaRequestHandler是由 KafkaRequestHandlerPool线程池进行管理的。与我们文章最开始提到的SocketServer创建相同,KafkaRequestHandlerPool也是在KafkaServer.startup()中创建的:

        /* start processing requests */
        apis = new KafkaApis(socketServer.requestChannel, replicaManager, groupCoordinator,
          kafkaController, zkUtils, config.brokerId, config, metadataCache, metrics, authorizer)
        //KafkaRequestHandlerPool线程池,用来管理所有KafkaRequestHandler线程
        requestHandlerPool = new KafkaRequestHandlerPool(config.brokerId, socketServer.requestChannel, apis, config.numIoThreads)

这里,KafkaApis是Kafka的API接口层,可以理解为一个工具类,职责就是解析请求然后获取请求类型,根据请求类型将请求交付给对应的业务层,可见,KafkaApis是实现了网络层到业务层的真正映射关系,下文会详解。

class KafkaRequestHandlerPool(val brokerId: Int,
                              val requestChannel: RequestChannel,
                              val apis: KafkaApis,
                              numThreads: Int) extends Logging with KafkaMetricsGroup {

  /* a meter to track the average free capacity of the request handlers */
  private val aggregateIdleMeter = newMeter("RequestHandlerAvgIdlePercent", "percent", TimeUnit.NANOSECONDS)

  this.logIdent = "[Kafka Request Handler on Broker " + brokerId + "], "
  val threads = new Array[Thread](numThreads)
  //初始化由KafkaRequestHandler线程构成的线程数组
  val runnables = new Array[KafkaRequestHandler](numThreads)
  for(i <- 0 until numThreads) {
    runnables(i) = new KafkaRequestHandler(i, brokerId, aggregateIdleMeter, numThreads, requestChannel, apis)
    threads(i) = Utils.daemonThread("kafka-request-handler-" + i, runnables(i))
    threads(i).start()
  }

KafkaRequestHandlerPool构造方法中初始化并启动了多个KafkaRequestHandler线程对象,线程池大小通过Kafka配置文件配置项num.io.threads进行配置。

KafkaRequestHandlerPool线程池中的所有KafkaRequestHandler,通过竞争方式从RequestChannel.requestQueue中获取请求进行处理。由于requestQueue的类型是ArrayBlockingQueue,通过调用ArrayBlockingQueue.poll()方法取出请求。我们看ArrayBlockingQueue.poll()方法源码,可以看到ArrayBlockingQueue.poll()方法线程安全,因此多个KafkaRequestHandler线程竞争requestQueue不会出现线程安全问题。
这是KafkaRequestHandler.run()方法,就是不断从requestQueue中取出请求,调用API层业务处理逻辑进行处理。

 def run() {
    while(true) {
      try {
        var req : RequestChannel.Request = null
        while (req == null) {
        //略
        req = requestChannel.receiveRequest(300)//从RequestChannel.requestQueue中取出请求
        //略
        apis.handle(req)//调用KafkaApi.handle(),将请求交付给业务
      } catch {}
    }
  }

6.KafkaApis

KafkaApis类似一个工具类,解析用户请求并将请求交付给业务层,我们可以把它看做Kafka的API层。从上面KafkaRequestHandler.run()方法可以看到,这是通过调用KafkaApis.handle()方法完成的。

 def handle(request: RequestChannel.Request) {
    try {
      ApiKeys.forId(request.requestId) match {
        case ApiKeys.PRODUCE => handleProducerRequest(request)
        case ApiKeys.FETCH => handleFetchRequest(request)
        case ApiKeys.LIST_OFFSETS => handleOffsetRequest(request)
        case ApiKeys.METADATA => handleTopicMetadataRequest(request)
        case ApiKeys.LEADER_AND_ISR => handleLeaderAndIsrRequest(request)
        //其它ApiKeys,略
      }
    } catch { //异常处理,略 }
    } finally{
      request.apiLocalCompleteTimeMs = SystemTime.milliseconds
      }
  }

通过Switch-Case代码块,根据请求中的requestId,将请求交付给handleProducerRequest()handleFetchRequest()等等handle*Request()方法。具体每个handle*Request()方法的处理流程不在本文描述范围之内,但是都是在获取返回结果以后,调用requestChannel.sendResponse()将response交付给RequestChannel。这里只是交付给RequestChannel保存在RequestChannel.responseQueues中。
在上文讲解Processor的时候说过,Procossor.processNewResponses()就是从requestChannel.responseQueues取出属于自己的连接上的响应,准备返回给客户端。


结束

这样,我们通过对Acceptor、Processor、RequestChannel、KafkaRequestHandler以及KafkaApis多个角色的解析,完成了整个Kafka的消息流通闭环,即从客户端建立连接、发送请求给Kafka Server进行处理、Kafka Server将请求交付给具体业务进行处理、业务将处理结果返回给网络层、网络层将结果通过NIO返回给客户端的整个流程。同时,由于多Processor线程、以及KafkaRequestHandlerPoll线程池的存在,通过交付-获取的方式而不是阻塞等待的方式,让整个消息处理实现完全的异步化,各个角色各司其职,模块之间无耦合,线程之间或者相互竞争任务,或者被上层安排处理部分任务,整个效率非常高,结构也相当清晰。Processor线程的数量、KafkaRequestHandlerPool线程池可配置,因此可以根据CPU以及内存性能,合理调整Kafka Server的并行程度和处理能力。对Kafka基于Reactor模式的网络层的理解,以及消息从网络层到业务层交付逻辑的理解,非常有利于我们对Kafka集群的管理以及对Kafka问题原因的排查诊断,同时,也可以基于各个不同的角色暴露出来的一个java metrics,我们可以对Kafka进行有效的调整优化。

es6写 react 用 相关内容

2017-08-21 17:12:52 libaineu2004 阅读数 2122

文章来源:http://www.cnblogs.com/binchen-china/p/5487795.html

上篇线程/进程并发服务器中提到,提高服务器性能在IO层需要关注两个地方,一个是文件描述符处理,一个是线程调度。

IO复用是什么?IO即Input/Output,在网络编程中,文件描述符就是一种IO操作。

为什么要IO复用?

1.网络编程中非常多函数是阻塞的,如connect,利用IO复用可以以非阻塞形式执行代码。

2.之前提到listen维护两个队列,完成握手的队列可能有多个就绪的描述符,IO复用可以批处理描述符。

3.有时候可能要同时处理TCP和UDP,同时监听多个端口,同时处理读写和连接等。

为什么epoll效率要比select高?

1.在连接数量较大的场景,select遍历需要每个描述符,epoll由内核维护事件表,只需要处理有响应的描述符。

2.select本身处理文件描述符受到限制,默认1024。

3.效率并不是绝对的,当连接率高,断开和连接频繁时,select不一定比epoll差。所以要根据具体场合使用。

epoll的两种模式,电平触发和边沿触发。

1.电平触发效率较边沿触发低,电平触发模式下,当epoll_wait返回的事件没有全部相应处理完毕,内核缓冲区还存在数据时,会反复通知,直到处理完成。epoll默认使用这种模式。

2.边沿触发效率较高,内核缓冲区事件只通知一次。

一个epoll实现demo

复制代码
  1 #include <iostream>
  2 #include <sys/socket.h>
  3 #include <sys/epoll.h>
  4 #include <netinet/in.h>
  5 #include <arpa/inet.h>
  6 #include <fcntl.h>
  7 #include <unistd.h>
  8 #include <stdio.h>  
  9 #include <stdlib.h>  
 10 #include <string.h>  
 11 #include <errno.h>
 12 
 13 using namespace std;
 14 
 15 #define MAXLINE 5
 16 #define OPEN_MAX 100
 17 #define LISTENQ 20
 18 #define SERV_PORT 5000
 19 #define INFTIM 1000
 20 
 21 int main(int argc, char* argv[])
 22 {
 23     int listen_fd, connfd_fd, socket_fd, epfd, nfds;
 24     ssize_t n;
 25     char line[MAXLINE];
 26     socklen_t clilen;
 27 
 28     //声明epoll_event结构体的变量,ev用于注册事件,数组用于回传要处理的事件
 29     struct epoll_event ev,events[20];
 30     //生成用于处理accept的epoll专用的文件描述符
 31     epfd=epoll_create(5);
 32     struct sockaddr_in clientaddr;
 33     struct sockaddr_in serveraddr;
 34     listen_fd = socket(AF_INET, SOCK_STREAM, 0);
 35     //设置与要处理的事件相关的文件描述符
 36     ev.data.fd = listen_fd;
 37     //设置要处理的事件类型
 38     ev.events=EPOLLIN|EPOLLET;
 39     //注册epoll事件
 40     epoll_ctl(epfd,EPOLL_CTL_ADD,listen_fd,&ev);
 41     
 42     memset(&serveraddr, 0, sizeof(serveraddr));  
 43     serveraddr.sin_family = AF_INET;  
 44     serveraddr.sin_addr.s_addr = htonl(INADDR_ANY);
 45     serveraddr.sin_port = htons(SERV_PORT);
 46     
 47     if (bind(listen_fd, (struct sockaddr*)&serveraddr, sizeof(serveraddr)) == -1)
 48     {  
 49         printf("bind socket error: %s(errno: %d)\n",strerror(errno),errno);  
 50         exit(0);  
 51     }  
 52     
 53     if (listen(listen_fd, LISTENQ) == -1)
 54     {  
 55         exit(0);  
 56     }   
 57     
 58     for ( ; ; ) 
 59     {
 60         //等待epoll事件的发生
 61         nfds = epoll_wait(epfd,events,20,500);
 62         //处理所发生的所有事件
 63         for (int i = 0; i < nfds; ++i)
 64         {
 65             if (events[i].data.fd == listen_fd)//如果新监测到一个SOCKET用户连接到了绑定的SOCKET端口,建立新的连接。
 66 
 67             {
 68                 connfd_fd = accept(listen_fd,(sockaddr *)&clientaddr, &clilen);
 69                 if (connfd_fd < 0){
 70                     perror("connfd_fd < 0");
 71                     exit(1);
 72                 }
 73                 char *str = inet_ntoa(clientaddr.sin_addr);
 74                 cout << "accapt a connection from " << str << endl;
 75                 //设置用于读操作的文件描述符
 76                 ev.data.fd = connfd_fd;
 77                 //设置用于注测的读操作事件
 78                 ev.events = EPOLLIN|EPOLLET;
 79                 //注册ev
 80                 epoll_ctl(epfd,EPOLL_CTL_ADD,connfd_fd,&ev);
 81             }
 82             else if (events[i].events&EPOLLIN)//如果是已经连接的用户,并且收到数据,那么进行读入。
 83             {            
 84                 memset(&line,'\0', sizeof(line));  
 85                 if ( (socket_fd = events[i].data.fd) < 0)
 86                     continue;
 87                 if ( (n = read(socket_fd, line, MAXLINE)) < 0) {
 88                     if (errno == ECONNRESET) {
 89                         close(socket_fd);
 90                         events[i].data.fd = -1;
 91                     } else
 92                         std::cout<<"readline error"<<std::endl;
 93                 } else if (n == 0) {
 94                     close(socket_fd);
 95                     events[i].data.fd = -1;
 96                 }                
 97                 cout << line << endl;
 98                 //设置用于写操作的文件描述符
 99                 ev.data.fd = socket_fd;
100                 //设置用于注测的写操作事件
101                 ev.events = EPOLLOUT|EPOLLET;
102                 //修改socket_fd上要处理的事件为EPOLLOUT
103                 //epoll_ctl(epfd,EPOLL_CTL_MOD,socket_fd,&ev);
104             }
105             else if (events[i].events&EPOLLOUT) // 如果有数据发送
106             {
107                 socket_fd = events[i].data.fd;
108                 write(socket_fd, line, n);
109                 //设置用于读操作的文件描述符
110                 ev.data.fd = socket_fd;
111                 //设置用于注测的读操作事件
112                 ev.events = EPOLLIN|EPOLLET;
113                 //修改socket_fd上要处理的事件为EPOLIN
114                 epoll_ctl(epfd,EPOLL_CTL_MOD,socket_fd,&ev);
115             }
116         }
117     }
118     return 0;
119 }
复制代码

 

执行效果如下:

QQ图片20160512223244

第一次学epoll时,容易错误的认为epoll也可以实现并发,其实正确的话是epoll可以实现高性能并发服务器,epoll只是提供了IO复用,在IO“并发”,真正的并发只能通过线程进程实现。

那为什么可以同时连接两个客户端呢?实际上这两个客户端都是在一个进程上运行的,前面提到过各个描述符之间是相互不影响的,所以是一个进程轮循在处理多个描述符。

Reactor模式:

Reactor模式实现非常简单,使用同步IO模型,即业务线程处理数据需要主动等待或询问,主要特点是利用epoll监听listen描述符是否有相应,及时将客户连接信息放于一个队列,epoll和队列都是在主进程/线程中,由子进程/线程来接管各个描述符,对描述符进行下一步操作,包括connect和数据读写。主程读写就绪事件。

大致流程图如下:

image

Preactor模式:

Preactor模式完全将IO处理和业务分离,使用异步IO模型,即内核完成数据处理后主动通知给应用处理,主进程/线程不仅要完成listen任务,还需要完成内核数据缓冲区的映射,直接将数据buff传递给业务线程,业务线程只需要处理业务逻辑即可。

大致流程如下:

image


es6写 react 用 相关内容

没有更多推荐了,返回首页