与 Spring 集成的 TCP 网关

Question

我需要实现一个 tcp 网关，将消息发送到服务器并同步接收响应。

服务器已启动并在特定端口上运行，因此按照我在网上找到的示例，我将其配置如下：

@Configuration
public class TcpClientConfig {

    @Value("${tcp.server.host}")
    private String host;

    @Value("${tcp.server.port}")
    private int port;

    private static final Logger LOGGER = LoggerFactory.getLogger(TcpClientConfig.class);


    @Component
    @MessagingGateway
    public interface TcpClientGateway {

        @Gateway(requestChannel="outboundChannel")
        byte[] sendByClient(byte[] message);
    }

    @Bean
    public AbstractClientConnectionFactory clientConnectionFactory() {
        TcpNioClientConnectionFactory tcpNioClientConnectionFactory = new TcpNioClientConnectionFactory(host, port);
        tcpNioClientConnectionFactory.setUsingDirectBuffers(true);
        tcpNioClientConnectionFactory.setSingleUse(false);
        Bytearraylengthheaderserializer serDeser = new Bytearraylengthheaderserializer(2);
        tcpNioClientConnectionFactory.setSerializer(serDeser);
        tcpNioClientConnectionFactory.setDeserializer(serDeser);
        return tcpNioClientConnectionFactory;
    }

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


    @Bean
    public IntegrationFlow incomingClient(final TcpReceivingChannelAdapter tcpReceivingChannelAdapter,
                                          TcpServerEndpoint tcpServerEndpoint) {

        return IntegrationFlows
                .from(tcpReceivingChannelAdapter)
                .handle(message -> { LOGGER.info("RECEIVING ON CLIENT: {}", tcpServerEndpoint.processMessage((byte[]) message.getPayload()));})
                .get();
    }


    @Bean
    public IntegrationFlow outgoingClient(final MessageChannel outboundChannel, final TcpSendingMessageHandler tcpSendingClientMessageHandler) {

        return IntegrationFlows
                .from(outboundChannel)
                .handle(tcpSendingClientMessageHandler)
                .get();
    }



    @Bean
    public TcpSendingMessageHandler tcpSendingClientMessageHandler(AbstractClientConnectionFactory clientConnectionFactory) {

        TcpSendingMessageHandler tcpSendingMessageHandler = new TcpSendingMessageHandler();
        tcpSendingMessageHandler.setConnectionFactory(clientConnectionFactory);
        tcpSendingMessageHandler.setClientMode(true);
        tcpSendingMessageHandler.setRetryInterval(5000);
        tcpSendingMessageHandler.setLoggingEnabled(true);
        return tcpSendingMessageHandler;
    }

    @Bean
    public TcpReceivingChannelAdapter tcpReceivingChannelAdapter(AbstractClientConnectionFactory clientConnectionFactory) {

        TcpReceivingChannelAdapter tcpReceivingChannelAdapter = new TcpReceivingChannelAdapter();
        tcpReceivingChannelAdapter.setConnectionFactory(clientConnectionFactory);
        tcpReceivingChannelAdapter.setAutoStartup(true);
        tcpReceivingChannelAdapter.setClientMode(true);
        tcpReceivingChannelAdapter.setRetryInterval(5000);
        return tcpReceivingChannelAdapter;
    }

    @EventListener
    public void open(TcpConnectionOpenEvent event)
            throws InterruptedException, IOException {
        LOGGER.info("Open new connection to Router {}", event.getConnectionId());
        clientConnectionFactory().getConnection().getSocketInfo().getChannel()
                .setOption(ExtendedSocketOptions.TCP_KEEPIDLE, 60);
    }

    @EventListener
    public void close(TcpConnectionCloseEvent event)
            throws InterruptedException, IOException {
        LOGGER.info("Close connection to Router {}", event.getConnectionId());
    }
}

这个想法是有一个网关来发送消息并同步等待响应。但是当我调用网关的方法时

byte[] sendByClient(byte[] message)

呼叫挂断，网关没有响应。

Answer 1

TcpSendingMessageHandler 是一种单向组件。它只是不等待任何回复在流程开始时为TcpClientGateway 生成replyChannel 标头。

考虑改用TcpOutboundGateway：https://docs.spring.io/spring-integration/docs/current/reference/html/ip.html#tcp-gateways

这个示例有一些基本的想法和如何使用网关：https://github.com/spring-projects/spring-integration-samples/tree/main/basic/tcp-client-server