Spring 集成使用事件侦听器处理连接关闭事件并在使用动态 TCP 路由时重新建立它

Question

我正在使用 spring 集成来创建请求/响应架构的流程，并从服务器接收任意数据。在此阶段之前，我检查了 spring-integration github 中的示例以及@Gary Russell 和@Artem Bilan 的建议。

这是我的网关界面

@Component
@MessagingGateway(defaultRequestChannel = "toTcp.input")
public interface ToTCP {
    byte[] send(String data, @Header("host") String host, @Header("port") int port, @Header("irregularMessageChannelName") String channelName);
    byte[] send(String data, @Header("host") String host, @Header("port") int port);
}

这是我的 TcpClientConfig

@Component
public class TcpClientConfig {
    @Bean
    public IntegrationFlow toTcp() {
        return f -> f.route(new TcpRouter());
    }
}

这是我的扩展 AbstractMessageRouter 的 TcpRouter

public class TcpRouter extends AbstractMessageRouter {


    private final Logger log = LoggerFactory.getLogger(TcpRouter.class);

    private final static int MAX_CACHED = 100; // When this is exceeded, we remove the LRU.

    private HashMap<String, Message<?>> connectionRegistery = new HashMap<>();

    private final LinkedHashMap<String, MessageChannel> subFlows =
        new LinkedHashMap<String, MessageChannel>(MAX_CACHED, .75f, true) {

            @Override
            protected boolean removeEldestEntry(Map.Entry<String, MessageChannel> eldest) {
                if (size() > MAX_CACHED) {
                    removeSubFlow(eldest);
                    return true;
                } else {
                    return false;
                }
            }

        };

    @Autowired
    private IntegrationFlowContext flowContext;

    @Override
    protected Collection<MessageChannel> determineTargetChannels(Message<?> message) {
        MessageChannel channel;
        boolean hasThisConnectionIrregularChannel = message.getHeaders().containsKey("irregularMessageChannelName");
        if (hasThisConnectionIrregularChannel) {
            channel = this.subFlows.get(message.getHeaders().get("host", String.class) + message.getHeaders().get("port") + ".extended");
        } else {
            channel = this.subFlows.get(message.getHeaders().get("host", String.class) + message.getHeaders().get("port"));
        }

        if (channel == null) {
            channel = createNewSubflow(message);
        }
        return Collections.singletonList(channel);
    }


    private MessageChannel createNewSubflow(Message<?> message) {
        String host = (String) message.getHeaders().get("host");
        Integer port = (Integer) message.getHeaders().get("port");

        boolean hasThisConnectionIrregularChannel = message.getHeaders().containsKey("irregularMessageChannelName");

        Assert.state(host != null && port != null, "host and/or port header missing");
        String flowRegisterKey;

        if (hasThisConnectionIrregularChannel) {
            flowRegisterKey = host + port + ".extended";
        } else {
            flowRegisterKey = host + port;
        }

        TcpNetClientConnectionFactory cf = new TcpNetClientConnectionFactory(host, port);
        cf.setSoTimeout(0);
        cf.setSoKeepAlive(true);

        ByteArrayCrLfSerializer byteArrayCrLfSerializer = new ByteArrayCrLfSerializer();
        byteArrayCrLfSerializer.setMaxMessageSize(1048576);

        cf.setSerializer(byteArrayCrLfSerializer);
        cf.setDeserializer(byteArrayCrLfSerializer);

        TcpOutboundGateway tcpOutboundGateway;
        if (hasThisConnectionIrregularChannel) {
            log.info("TcpRouter # createNewSubflow extended TcpOutboundGateway will be created");
            String irregularMessageChannelName = (String) message.getHeaders().get("irregularMessageChannelName");
            DirectChannel directChannel = getBeanFactory().getBean(irregularMessageChannelName, DirectChannel.class);
            tcpOutboundGateway = new ExtendedTcpOutboundGateway(directChannel);
        } else {
            log.info("TcpRouter # createNewSubflow extended TcpOutboundGateway will be created");
            tcpOutboundGateway = new TcpOutboundGateway();
        }

        tcpOutboundGateway.setConnectionFactory(cf);

        tcpOutboundGateway.setAdviceChain(Arrays.asList(new Advice[]{tcpRetryAdvice()}));

        IntegrationFlow flow = f -> f.handle(tcpOutboundGateway);

        IntegrationFlowContext.IntegrationFlowRegistration flowRegistration =
            this.flowContext.registration(flow)
                //.addBean(cf)
                .addBean("client_connection_" + flowRegisterKey, cf)
                .id(flowRegisterKey + ".flow")
                .register();

        MessageChannel inputChannel = flowRegistration.getInputChannel();

        this.subFlows.put(flowRegisterKey, inputChannel);
        this.connectionRegistery.put("client_connection_" + flowRegisterKey, message);

        return inputChannel;
    }

    private void removeSubFlow(Map.Entry<String, MessageChannel> eldest) {
        String hostPort = eldest.getKey();
        this.flowContext.remove(hostPort + ".flow");
    }

    @Bean
    public RequestHandlerRetryAdvice tcpRetryAdvice() {
        SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();
        retryPolicy.setMaxAttempts(3);

        ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
        backOffPolicy.setInitialInterval(100);
        backOffPolicy.setMaxInterval(1000);
        backOffPolicy.setMultiplier(2);

        RetryTemplate retryTemplate = new RetryTemplate();
        retryTemplate.setRetryPolicy(retryPolicy);
        retryTemplate.setBackOffPolicy(backOffPolicy);

        RequestHandlerRetryAdvice tcpRetryAdvice = new RequestHandlerRetryAdvice();
        tcpRetryAdvice.setRetryTemplate(retryTemplate);

        // This allows fail-controlling
        tcpRetryAdvice.setRecoveryCallback(new ErrorMessageSendingRecoverer(failMessageChannel()));

        return tcpRetryAdvice;
    }

    @Bean
    public MessageChannel failMessageChannel() {
        return new DirectChannel();
    }

    @ServiceActivator(inputChannel = "failMessageChannel")
    public void messageAggregation(String in) {
        log.error("TcpRouter # connection retry failed with message : " + in);
    }

    @Autowired
    private ToTCP toTCP;

    @EventListener
    public void listen(TcpConnectionCloseEvent event) {
        String connectionFactoryName = event.getConnectionFactoryName();
        boolean isConnectionRegistered = this.connectionRegistery.containsKey(connectionFactoryName);
        if (isConnectionRegistered) {
            Message<?> message = this.connectionRegistery.get(connectionFactoryName);
            String host = (String) message.getHeaders().get("host");
            Integer port = (Integer) message.getHeaders().get("port");
            boolean hasThisConnectionIrregularChannel = message.getHeaders().containsKey("irregularMessageChannelName");
            if (hasThisConnectionIrregularChannel) {
                log.info("TcpRouter # listen # registered tcp connection with arbitrary message channel closed for host {} and port {}, it will open again !!", host, port);
                String unsolicitedMessageChannelName = (String) message.getHeaders().get("irregularMessageChannelName");
                toTCP.send(message.getPayload().toString(), host, port, unsolicitedMessageChannelName);
            } else {
                log.info("TcpRouter # listen # registered tcp connection closed for host {} and port {}, it will open again !!", host, port);
                toTCP.send(message.getPayload().toString(), host, port);
                            }
        } else {
            log.info("TcpRouter # listen # unregistered tcp connection closed, no action required.");
        }
    }
}

如果有任何连接关闭事件，我可以用事件监听器来处理。在事件侦听器中，我可以从 addBean("client_connection_" + flowRegisterKey, cf) 中注册的 connectionFactoryName 了解。这是该部分的

处理关闭连接后，我应该再次打开它以继续接收任意数据或准备好TCP服务器之间的连接以发送任何请求...但我是不确定我重新建立连接与发送数据的方式。

我应该使用

@Autowired
private ToTCP toTCP;

在 TcpRouter 中 class 再次发送消息

或

我应该直接发消息给

@Override
protected Collection<MessageChannel> determineTargetChannels(Message<?> message)

方法。我对他们的工作行为感到困惑......你能给我正确的想法，帮助我使用更方便的方式让 EventListener 重新建立连接吗？

Answer 1

Actually you are right, reconnection request is same with initial time i called it.

Should i use determineTargetChannels in that case ?

否；在事件侦听器中执行与首先调用 ToTCP 完全相同的操作（发送新请求并处理回复）。

Spring 集成使用事件侦听器处理连接关闭事件并在使用动态 TCP 路由时重新建立它

Spring integration Handle Connection Close event with Event Listener and Re establish it while using Dynamic TCP Routing

java

spring

tcp

spring-integration

spring-boot