Wie wird eine Keep Alive-Verbindung auf einer TCP-Verbindung mithilfe der Federintegration implementiert?

Question

Ich habe einen TCP-Client, der mit Spring Integration TCP gebaut wurde und der Server unterstützt eine Keep Alive-Nachricht (Ping/Pong-Stil). Die Verbindungen wurden mit einer CachingClientConnectionFactory konfiguriert und ich möchte diese Serverfunktion nutzen. Hier ist meine Bean-Konfiguration:

private static final int SERIALIZER_HEADER_SIZE = 2; 

/** 
* Serializer used by connection factory to send and receive messages 
*/ 
@Bean 
public ByteArrayLengthHeaderSerializer byteArrayLengthHeaderSerializer() { 
    return new ByteArrayLengthHeaderSerializer(SERIALIZER_HEADER_SIZE); 
} 

@Bean 
public AbstractClientConnectionFactory tcpClientConnectionFactory() { 
    TcpNetClientConnectionFactory connFactory = 
     new TcpNetClientConnectionFactory(props.getUrl(), props.getPort()); 
    connFactory.setSerializer(byteArrayLengthHeaderSerializer()); 
    connFactory.setDeserializer(byteArrayLengthHeaderSerializer()); 
    connFactory.setSoTimeout(props.getSoTimeout()); 
    if (props.isUseSSL()) { 
     connFactory.setTcpSocketFactorySupport(new DefaultTcpNetSSLSocketFactorySupport(() -> { 
      return SSLContext.getDefault(); 
     })); 
    } 

    return connFactory; 
} 

/** 
* Connection factory used to create TCP client socket connections 
*/ 
@Bean 
public AbstractClientConnectionFactory tcpCachedClientConnectionFactory() { 
    CachingClientConnectionFactory cachingConnFactory = 
     new CachingClientConnectionFactory(tcpClientConnectionFactory(), props.getMaxPoolSize()); 
    cachingConnFactory.setConnectionWaitTimeout(props.getMaxPoolWait()); 
    return cachingConnFactory; 
} 

die Lösung hier Configure keep alive to keep connection alive all the time geschrieben Verwendung von I die Verbindung geöffnet halten kann, aber ich wollte auch Einfluss auf diesem Server nehmen Alive-Nachrichten halten und diese Nachrichten von Zeit zu Zeit zu überprüfen, zu senden, wenn die Verbindung ist noch am Leben. Dies kann die Leistung auf der Clientseite verbessern, da keine neue Verbindung neu hergestellt/erstellt werden muss, wenn der Socket geschlossen wurde.

Basierend darauf hat jemand einen Vorschlag, wie man dies mithilfe der Federintegration implementieren kann?

Answer 1

Bei Verwendung einer einfachen Clientverbindungsfactory ist es einfach, Heartbeat-Nachrichten auf Anwendungsebene mit einer @InboundChannelAdapter einzurichten.

Einfaches Beispiel:

@SpringBootApplication 
public class So46918267Application { 

    public static void main(String[] args) throws IOException { 
     // Simulated Server 
     final ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(1234); 
     ExecutorService executor = Executors.newSingleThreadExecutor(); 
     executor.execute(() -> { 
      try { 
       Socket socket = server.accept(); 
       BufferedReader reader = new BufferedReader(new InputStreamReader(socket.getInputStream())); 
       String line; 
       while ((line = reader.readLine()) != null) { 
        System.out.println(line); 
        if (line.equals("keep_alive")) { 
         socket.getOutputStream().write("OK\r\n".getBytes()); 
        } 
       } 
      } 
      catch (IOException e) { 
       e.printStackTrace(); 
      } 
     }); 
     ConfigurableApplicationContext context = SpringApplication.run(So46918267Application.class, args); 
     System.out.println("Hit enter to terminate"); 
     System.in.read(); 
     executor.shutdownNow(); 
     context.close(); 
     server.close(); 
    } 

    @Bean 
    public TcpNetClientConnectionFactory client() { 
     return new TcpNetClientConnectionFactory("localhost", 1234); 
    } 

    @ServiceActivator(inputChannel = "toTcp") 
    @Bean 
    public TcpOutboundGateway gateway() { 
     TcpOutboundGateway gateway = new TcpOutboundGateway(); 
     gateway.setConnectionFactory(client()); 
     return gateway; 
    } 

    // HEARTBEATS 

    private final Message<?> heartbeatMessage = MessageBuilder.withPayload("keep_alive") 
      .setReplyChannelName("heartbeatReplies") 
      .build(); 

    @InboundChannelAdapter(channel = "toTcp", poller = @Poller(fixedDelay = "25000")) 
    public Message<?> heartbeat() { 
     return this.heartbeatMessage; 
    } 

    @ServiceActivator(inputChannel = "heartbeatReplies") 
    public void reply(byte[] reply) { 
     System.out.println(new String(reply)); 
    } 

} 

Wenn die CachingClientConnectionFactory verwenden, aber es ist nicht klar, warum man einen Pool von Verbindungen im Leerlauf offen halten möchte. Die Funktionsweise des Pools besteht jedoch darin, dass die inaktiven Verbindungen in einer Warteschlange verbleiben, sodass jede Anforderung an die älteste Verbindung gesendet und die Verbindung an das Ende der Warteschlange zurückgegeben wird.

maxMessagesPerPoll fügte hinzu, dass Anzahl der Nachrichten an jeder Umfrage emittieren würde und ...

@InboundChannelAdapter(channel = "toTcp", 
    poller = @Poller(fixedDelay = "25000", maxMessagesPerPoll = "5")) 

offen für 5 Verbindungen halten würde. Es werden keine neuen Verbindungen geöffnet (wenn es mindestens eine gibt), aber wenn der Pool 5 oder mehr Verbindungen enthält, werden mindestens 5 geöffnet bleiben. Wenn keine offenen Verbindungen vorhanden sind, wird nur eine Verbindung geöffnet.