Getty-Detaillierte Erläuterung der Implementierung des Java NIO-Framework-Designs

Vorwort

Getty ist ein NIO-Framework, das ich geschrieben habe, um Java NIO zu lernen. Während des Implementierungsprozesses habe ich mich auf das Design von Netty bezogen und es mit Groovy implementiert. Obwohl es sich nur um ein Spielzeug handelt, ist der Spatz klein und verfügt über alle inneren Organe. Während des Implementierungsprozesses habe ich mich nicht nur mit der Verwendung von NIO vertraut gemacht, sondern auch viele Designideen von Netty gelernt und so meine Codierungs- und Designfähigkeiten verbessert .

Was den Grund betrifft, warum ich Groovy zum Schreiben verwende, weil ich Groovy gerade erst zum Üben verwendet habe. Außerdem ist Groovy mit Java kompatibel, sodass der Unterschied nur in der Syntax liegt basierend auf Java API.

Gettys Kerncodezeilen überschreiten nicht 500 Zeilen. Einerseits profitiert es von der prägnanten Syntax von Groovy und andererseits, weil ich nur die Kernlogik implementiert habe Durchführung. Ein Gerüst ist einfach zu bauen, aber der Bau eines Wolkenkratzers ist nicht so einfach, aber zum Erlernen von NIO reicht es aus.

ThreadedModell

Getty verwendet Reactoder ein Multithread-Modell

1. Es gibt einen dedizierten NIO-Thread – der Acceptor-Thread wird verwendet, um den Server zu überwachen, die TCP-Verbindungsanforderung des Clients zu empfangen und dann die Verbindung dem Worker-Thread zuzuweisen, der die Lese- und Schreibereignisse überwacht .

2. Netzwerk-IO-Operationen – Lesen/Schreiben usw. – sind für mehrere Arbeitsthreads verantwortlich, die für das Lesen, Dekodieren, Kodieren und Senden von Nachrichten verantwortlich sind.

3. 1 Arbeitsthread kann N Links gleichzeitig verarbeiten, aber 1 Link entspricht nur 1 Arbeitsthread, um gleichzeitige Betriebsprobleme zu vermeiden.

Ereignisgesteuertes Modell

Die gesamte serverseitige Prozessverarbeitung basiert auf dem Ereignismechanismus. Im Prozess [Verbindung akzeptieren -> Lesen -> Geschäftsverarbeitung -> Schreiben -> Verbindung schließen] löst der Trigger das entsprechende Ereignis aus und der Ereignishandler verarbeitet das entsprechendes Ereignis. Reaktion zum Abschluss der serverseitigen Geschäftsverarbeitung.

Ereignisdefinition

1. onRead: Dieses Ereignis wird ausgelöst, wenn der Client Daten sendet und vom Arbeitsthread korrekt gelesen wurde. Dieses Ereignis benachrichtigt jeden Event-Handler darüber, dass die vom Client gesendeten Daten tatsächlich verarbeitet werden können.

2. onWrite: Dieses Ereignis wird ausgelöst, wenn der Client beginnen kann, vom Server gesendete Daten zu akzeptieren. Durch dieses Ereignis können wir Antwortdaten an den Client senden. (Das Schreibereignis wird in der aktuellen Implementierung nicht verwendet)

3. onClosed: Dieses Ereignis wird ausgelöst, wenn der Client die Verbindung zum Server trennt.

Implementierung des Ereignisrückrufmechanismus

In diesem Modell werden Ereignisse übertragen, d. h. alle registrierten Ereignishandler können Ereignisbenachrichtigungen erhalten. Auf diese Weise können Geschäftsabwicklungen unterschiedlicher Art mit unterschiedlichen Prozessoren implementiert werden, wobei die Funktion jedes Prozessors möglichst einheitlich ist.

Wie unten gezeigt: Das gesamte Ereignismodell besteht aus Listenern, Ereignisadaptern, Ereignisauslösern (HandlerChain, PipeLine) und Ereignisprozessoren.

• ServerListener: Dies ist eine Ereignis-Schnittstelle , die die zu überwachenden Serverereignisse

  interface ServerListener extends Serializable{
      /**
       * 可读事件回调
       * @param request
       */
      void onRead(ctx)
      /**
       * 可写事件回调
       * @param request
       * @param response
       */
      void onWrite(ctx)
      /**
       * 连接关闭回调
       * @param request
       */
      void onClosed(ctx)
  }

  definiert

• EventAdapter: Implementieren Sie einen Adapter (EventAdapter) für die Serverlistener-Schnittstelle. Der Vorteil davon besteht darin, dass der endgültige Ereignishandler nur die betreffenden Ereignisse verarbeiten kann.

  class EventAdapter implements ServerListener {
      //下个处理器的引用
      protected next
      void onRead(Object ctx) {
      }
      void onWrite(Object ctx) {
      }
      void onClosed(Object ctx) {
      }
  }

• Nicht<a href="http://www.php.cn/wiki/109.html" target="_blank">wenn<code>Not<a href="http://www.php.cn/wiki/109.html" target="_blank">if</a>ier ier: Wird verwendet, um registrierte Ereignishandler zu benachrichtigen, auf Ereignisse zu reagieren, indem sie zu geeigneten Zeiten Serverereignisse auslösen.

  interface Notifier extends Serializable{
      /**
       * 触发所有可读事件回调
       */
      void fireOnRead(ctx)
      /**
       * 触发所有可写事件回调
       */
      void fireOnWrite(ctx)
      /**
       * 触发所有连接关闭事件回调
       */
      void fireOnClosed(ctx)
  }

• HandlerChain: Implementiert die Notifier-Schnittstelle, um eine geordnete Ereignishandlerkette aufrechtzuerhalten, die jedes Mal vom ersten Handler aus ausgelöst wird.

  class HandlerChain implements Notifier{
      EventAdapter head
      EventAdapter tail
      /**
       * 添加处理器到执行链的最后
       * @param handler
       */
      void addLast(handler) {
          if (tail != null) {
              tail.next = handler
              tail = tail.next
          } else {
              head = handler
              tail = head
          }
      }
      void fireOnRead(ctx) {
          head.onRead(ctx)
      }
      void fireOnWrite(ctx) {
          head.onWrite(ctx)
      }
      void fireOnClosed(ctx) {
          head.onClosed(ctx)
      }
  }

• PipeLine: Implementiert die Notifier-Schnittstelle als Ereignisbus, um eine Liste von Ereignisketten zu verwalten.

  class PipeLine implements Notifier{
      static logger = LoggerFactory.getLogger(PipeLine.name)
      //监听器队列
      def listOfChain = []
      PipeLine(){}
      /**
       * 添加监听器到监听队列中
       * @param chain
       */
      void addChain(chain) {
          synchronized (listOfChain) {
              if (!listOfChain.contains(chain)) {
                  listOfChain.add(chain)
              }
          }
      }
      /**
       * 触发所有可读事件回调
       */
      void fireOnRead(ctx) {
          logger.debug("fireOnRead")
          listOfChain.each { chain ->
              chain.fireOnRead(ctx)
          }
      }
      /**
       * 触发所有可写事件回调
       */
      void fireOnWrite(ctx) {
          listOfChain.each { chain ->
              chain.fireOnWrite(ctx)
          }
      }
      /**
       * 触发所有连接关闭事件回调
       */
      void fireOnClosed(ctx) {
          listOfChain.each { chain ->
              chain.fireOnClosed(ctx)
          }
      }
  }

Ereignisverarbeitungsprozess

Programmiermodell

Die Ereignisverarbeitung übernimmt das Chain-of-Responsibility-Modell. Nachdem der Prozessor die Daten verarbeitet hat, entscheidet er, ob er mit der Ausführung des nächsten Prozessors fortfährt. Wenn der Prozessor die Aufgabe nicht zur Verarbeitung an den Thread-Pool übergibt, wird der gesamte Verarbeitungsprozess im selben Thread verarbeitet. Und jede Verbindung hat ein separates PipeLine, und der Arbeitsthread kann zwischen mehreren Verbindungskontexten wechseln, aber ein Verbindungskontext wird nur von einem Thread verarbeitet.

核心类

ConnectionCtx

连接上下文ConnectionCtx

class ConnectionCtx {
    /**socket连接*/
    SocketChannel channel
    /**用于携带额外参数*/
    Object attachment
    /**处理当前连接的工作线程*/
    Worker worker
    /**连接超时时间*/
    Long timeout
    /**每个连接拥有自己的pipeline*/
    PipeLine pipeLine
}

NioServer

主线程负责监听端口，持有工作线程的引用（使用轮转法分配连接），每次有连接到来时，将连接放入工作线程的连接队列，并唤醒线程selector.wakeup()（线程可能阻塞在selector上）。

class NioServer extends Thread {
    /**服务端的套接字通道*/
    ServerSocketChannel ssc
    /**选择器*/
    Selector selector
    /**事件总线*/
    PipeLine pipeLine
    /**工作线程列表*/
    def workers = []
    /**当前工作线程索引*/
    int index
}

Worker

工作线程，负责注册server传递过来的socket连接。主要监听读事件，管理socket，处理写操作。

class Worker extends Thread {
    /**选择器*/
    Selector selector
    /**读缓冲区*/
    ByteBuffer buffer
    /**主线程分配的连接队列*/
    def queue = []
    /**存储按超时时间从小到大的连接*/
    TreeMap<Long, ConnectionCtx> ctxTreeMap

    void run() {
        while (true) {
            selector.select()
            //注册主线程发送过来的连接
            registerCtx()
            //关闭超时的连接
            closeTimeoutCtx()
            //处理事件
            dispatchEvent()
        }
    }
}

运行一个简单的 Web 服务器

我实现了一系列处理HTTP请求的处理器，具体实现看代码。

• LineBasedDecoder：行解码器，按行解析数据

• HttpRequestDecoder：HTTP请求解析，目前只支持GET请求

• HttpRequestHandler：Http 请求处理器，目前只支持GET方法

• HttpResponseHandler：Http响应处理器

下面是写在test中的例子

class WebServerTest {
    static void main(args) {
        def pipeLine = new PipeLine()

        def readChain = new HandlerChain()
        readChain.addLast(new LineBasedDecoder())
        readChain.addLast(new HttpRequestDecoder())
        readChain.addLast(new HttpRequestHandler())
        readChain.addLast(new HttpResponseHandler())

        def closeChain = new HandlerChain()
        closeChain.addLast(new ClosedHandler())

        pipeLine.addChain(readChain)
        pipeLine.addChain(closeChain)

        NioServer nioServer = new NioServer(pipeLine)
        nioServer.start()
    }
}

另外，还可以使用配置文件getty.properties设置程序的运行参数。

#用于拼接消息时使用的二进制数组的缓存区
common_buffer_size=1024
#工作线程读取tcp数据的缓存大小
worker_rcv_buffer_size=1024
#监听的端口
port=4399
#工作线程的数量
worker_num=1
#连接超时自动断开时间
timeout=900
#根目录
root=.

总结

Getty是我造的第二个小轮子，第一个是RedisHttpSession。都说不要重复造轮子。这话我是认同的，但是掌握一门技术最好的方法就是实践，在没有合适项目可以使用新技术的时候，造一个简单的轮子是不错的实践手段。

Getty 的缺点或者说还可以优化的点：

1. 线程的使用直接用了Thread类，看起来有点low。等以后水平提升了再来抽象一下。

2. 目前只有读事件是异步的，写事件是同步的。未来将写事件也改为异步的。

Das obige ist der detaillierte Inhalt vonGetty-Detaillierte Erläuterung der Implementierung des Java NIO-Framework-Designs. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!