Getty-Explication détaillée de la mise en œuvre de la conception du framework Java NIO-javaDidacticiel-php.cn

Avant-propos

Getty est un framework NIO que j'ai écrit pour apprendre Java NIO Au cours du processus d'implémentation, j'ai fait référence à la conception de Netty et j'ai utilisé Groovy pour l'implémenter. Bien qu'il ne s'agisse que d'un jouet, le moineau est petit et possède tous les organes internes. Au cours du processus de mise en œuvre, je me suis non seulement familiarisé avec l'utilisation de NIO, mais j'ai également appris de nombreuses idées de conception de Netty, améliorant ainsi mes capacités de codage et de conception. .

Quant à la raison pour laquelle j'utilise Groovy pour l'écrire, parce que je viens d'apprendre Groovy et que je l'ai utilisé pour m'entraîner. De plus, Groovy est compatible avec Java, donc la différence réside uniquement dans la syntaxe. basé sur Java API.

Les lignes de code de base de Getty ne dépassent pas 500 lignes. D'une part, il bénéficie de la syntaxe concise de Groovy, et d'autre part, parce que je n'ai implémenté que la logique de base. mise en œuvre. Les échafaudages sont faciles à construire, mais construire un gratte-ciel n'est pas si simple, mais c'est suffisant pour apprendre NIO.

Modèlethreadé

Getty utilise Reactou un modèle multithread

Getty-Explication détaillée de la mise en œuvre de la conception du framework Java NIO

Il existe un thread NIO dédié - le thread Acceptor est utilisé pour surveiller le serveur, recevoir la demande de connexion TCP du client, puis attribuer la connexion au thread de travail, qui surveille les événements de lecture et d'écriture .
Les opérations d'E/S réseau - lecture/écriture, etc. sont responsables de plusieurs threads de travail, qui sont responsables de la lecture, du décodage, de l'encodage et de l'envoi des messages.
1 thread de travail peut traiter N liens en même temps, mais 1 lien ne correspond qu'à 1 thread de travail pour éviter les problèmes de fonctionnement simultané.

Modèle basé sur les événements

L'ensemble du traitement du processus côté serveur est basé sur le mécanisme d'événements. Dans le processus [Accepter la connexion -> Lecture -> Traitement métier -> Écriture -> Fermer la connexion], le déclencheur déclenchera l'événement correspondant et le gestionnaire d'événements gérera le événement correspondant respectivement. Réponse pour terminer le traitement commercial côté serveur.

Définition de l'événement

onRead : Cet événement est déclenché lorsque le client envoie des données et a été correctement lu par le thread de travail. Cet événement informe chaque gestionnaire d'événements que les données envoyées par le client peuvent réellement être traitées.
onWrite : Cet événement est déclenché lorsque le client peut commencer à accepter les données envoyées par le serveur. Grâce à cet événement, nous pouvons envoyer des données de réponse au client. (L'événement write n'est pas utilisé dans l'implémentation actuelle)
onClosed : Cet événement est déclenché lorsque le client se déconnecte du serveur.

Mise en œuvre d'un mécanisme de rappel d'événements

Dans ce modèle, les événements sont diffusés, c'est-à-dire que tous les gestionnaires d'événements enregistrés peuvent recevoir des notifications d'événements. De cette manière, des traitements métier de différentes natures peuvent être mis en œuvre à l'aide de différents processeurs, rendant la fonction de chaque processeur aussi unique que possible.

Comme indiqué ci-dessous : l'ensemble du modèle d'événement se compose d'écouteurs, d'adaptateurs d'événements, de déclencheurs d'événements (HandlerChain, PipeLine) et de processeurs d'événements.

Getty-Explication détaillée de la mise en œuvre de la conception du framework Java NIO

ServerListener : Il s'agit d'une interface d'événement qui définit les événements du serveur à surveiller

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

Copier après la connexion

EventAdapter : Implémentez un adaptateur (EventAdapter) pour l'interface Serverlistener. L'avantage est que le gestionnaire d'événements final ne peut gérer que les événements concernés.
```
class EventAdapter implements ServerListener {
    //下个处理器的引用
    protected next
    void onRead(Object ctx) {
    }
    void onWrite(Object ctx) {
    }
    void onClosed(Object ctx) {
    }
}
```
Copier après la connexion
Pas<a href="http://www.php.cn/wiki/109.html" target="_blank">si<code>Not<a href="http://www.php.cn/wiki/109.html" target="_blank">if</a>ier ier : utilisé pour avertir les gestionnaires d'événements enregistrés de répondre aux événements en déclenchant des événements du serveur à des moments appropriés.
```
interface Notifier extends Serializable{
    /**
     * 触发所有可读事件回调
     */
    void fireOnRead(ctx)
    /**
     * 触发所有可写事件回调
     */
    void fireOnWrite(ctx)
    /**
     * 触发所有连接关闭事件回调
     */
    void fireOnClosed(ctx)
}
```
Copier après la connexion

HandlerChain : implémente l'interface Notifier pour maintenir une chaîne de gestionnaires d'événements ordonnée, se déclenchant à partir du premier gestionnaire à chaque fois.

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)
    }
}

Copier après la connexion

PipeLine : implémente l'interface Notifier en tant que bus d'événements pour maintenir une liste de chaînes d'événements.

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)
        }
    }
}

Copier après la connexion

Processus de traitement des événements

Getty-Explication détaillée de la mise en œuvre de la conception du framework Java NIO

Modèle de programmation

Le traitement des événements adopte le modèle de chaîne de responsabilité, chacun Une fois que le processeur a traité les données, il décidera s'il doit continuer à exécuter le processeur suivant. Si le processeur ne transmet pas la tâche au pool de threads pour traitement, l'ensemble du processus de traitement est traité dans le même thread. Et chaque connexion a un PipeLine distinct, et le thread de travail peut basculer entre plusieurs contextes de connexion, mais un contexte de connexion ne sera traité que par un seul thread.

核心类

ConnectionCtx

连接上下文ConnectionCtx

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

Copier après la connexion

NioServer

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

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

Copier après la connexion

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()
        }
    }
}

Copier après la connexion

运行一个简单的 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()
    }
}

Copier après la connexion

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

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

Copier après la connexion