带垂直的QuickFIX/J服务器。十、

Question

我正在尝试创建一个启动QuickFIX/J接受器服务器（TCP FIX服务器）的版本。启动时，接受器线程在单独的线程上运行，Vert. x不知道这一点（不会阻止事件循环）。然而，我可以从接受器线程访问事件总线，并将消息传递给其他顶点。

问题是，这是一个好的做法吗？

package com.millenniumit.fixgateway.service.impl.quickfix;

import io.vertx.core.AbstractVerticle;
import io.vertx.core.Promise;
import io.vertx.core.eventbus.MessageConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import quickfix.*;
import quickfix.fix42.NewOrderSingle;

public class FIXServerVerticle extends AbstractVerticle {

    private DynamicSessionProviderConfigHelper dynamicSessionProviderConfig;
    private ThreadedSocketAcceptor threadedSocketAcceptor;
    private static final Logger LOGGER = LoggerFactory.getLogger(FIXServerVerticle.class);

    /**
     * You can’t block waiting for the tcp server to bind in the start method as that would break the Golden Rule.
     * To prevent this, implement the asynchronous start method. This version of the method takes a Future as a parameter.
     * When the method returns the verticle will not be considered deployed.
     * @param startPromise
     */
    @Override
    public void start(Promise<Void> startPromise) {
        Application serverApplication = new Application() {
            @Override
            public void onCreate(SessionID sessionID) {
                LOGGER.info("Session Created : " + sessionID);
            }

            @Override
            public void onLogon(SessionID sessionID) {

            }

            @Override
            public void onLogout(SessionID sessionID) {

            }

            @Override
            public void toAdmin(Message message, SessionID sessionID) {

            }

            @Override
            public void fromAdmin(Message message, SessionID sessionID) throws FieldNotFound, IncorrectDataFormat, IncorrectTagValue, RejectLogon {

            }

            @Override
            public void toApp(Message message, SessionID sessionID) throws DoNotSend {

            }

            @Override
            public void fromApp(Message message, SessionID sessionID) throws FieldNotFound, IncorrectDataFormat, IncorrectTagValue, UnsupportedMessageType {
                LOGGER.info("Processing in worker thread: " + message);
                //Offload processing logic from event loop
                getVertx().executeBlocking(future -> {
                    //blocking code, run on the worker thread
                    LOGGER.info("Processing in worker thread: " + message);
                    //processing logic
                    future.complete(message);
                }, res -> {
                    //non blocking code running on the event loop thread
                    getVertx().eventBus().request("in.message", res.result(), ar -> {
                        if (ar.succeeded()) {
                            Session.lookupSession(sessionID).send((Message) ar.result().body());
                        }
                    });
                });
            }
        };
        //Offload acceptor initialization from event loop
        getVertx().executeBlocking(future -> {
            //blocking code, run on the worker thread
            MessageStoreFactory messageStoreFactory = new NoopStoreFactory();
            MessageFactory messageFactory = new DefaultMessageFactory();
            dynamicSessionProviderConfig = new DynamicSessionProviderConfigHelper();
            try {
                SessionSettings sessionSettings = new SessionSettings("acceptor-config");
                threadedSocketAcceptor = new ThreadedSocketAcceptor(serverApplication, messageStoreFactory, sessionSettings, messageFactory);
                dynamicSessionProviderConfig.configure(threadedSocketAcceptor, serverApplication, messageStoreFactory, sessionSettings, messageFactory);
                threadedSocketAcceptor.start();
                future.complete();
            } catch (ConfigError | FieldConvertError configError) {
                configError.printStackTrace();
                future.fail(configError.getMessage());
            }
        }, res -> {
            //non blocking code running on the event loop thread
            if(res.succeeded()){
                startPromise.complete();
            }else{
                startPromise.fail(res.cause().getMessage());
            }
        });
    }

    public void stop(Promise<Void> stopPromise) {
        //Offload acceptor stop method from event loop
        getVertx().executeBlocking(future -> {
            //blocking code, run on the worker thread
            threadedSocketAcceptor.stop();
            future.complete();
        }, res -> {
            //non blocking code running on the event loop thread
            if(res.succeeded()){
                stopPromise.complete();
            }else{
                stopPromise.fail(res.cause().getMessage());
            }
        });
    }

}

Answer 1

这种方法在我看来很好，我认为与vert. x的工作方式没有冲突。

顺便说一句，你可以稍微简化你的代码。

这个：

getVertx().executeBlocking(
  future -> {},
  res -> {
            if(res.succeeded()){
                startPromise.complete();
            }else{
                startPromise.fail(res.cause().getMessage());
            }
        }
)

可替换为：

getVertx().executeBlocking(
  future -> {},
  startPromise
)

Answer 2

我对vert不熟悉。x、 但是，一般来说，如果需要更高的吞吐量，QuickFIX/J应用程序通常会将传入消息卸载到单独的线程/队列。