Spring 集成 TCP 客户端 - 客户端不接收多条消息。需要处理onMessage事件答案

【问题标题】：Spring integrated TCP client - Client not receiving multiple messages. Need to handle onMessage eventSpring 集成 TCP 客户端 - 客户端不接收多条消息。需要处理onMessage事件
【发布时间】：2020-12-29 19:33:45
【问题描述】：

嗯，我有一个集成了 Spring 的 TCP 客户端。我正在尝试连接到远程 TCP 服务器并从服务器异步写入的套接字接收数据。

但是，碰巧的是，我的客户端正在接收第一条消息，并且没有进一步从服务器套接字接收更多消息。（实际上查看服务器日志可以确定客户端已失去连接，但为什么？）

还有一件事是我如何在收到消息的那一刻触发某些功能？ - 可能是 TcpConnectionHandler-handleMessage() 或 TcpLisetner - onMessage()。

最后，我想要一个 Spring TCP 客户端，它连接到远程服务器并接收数据，因为它到达了套接字。以下是我的配置和代码：

我的配置：

<bean id="javaSerializer" class="com.my.client.CustomSerializerDeserializer" />
<bean id="javaDeserializer" class="com.my.client.CustomSerializerDeserializer" />

<context:property-placeholder />

<!-- Client side -->

<int:gateway id="gw" service-interface="com.zebra.client.SimpleGateway" default-request-channel="input" default-reply-channel="replies" />

<int-ip:tcp-connection-factory id="client"
    type="client" host="localhost" port="5678" single-use="false"
    so-timeout="100000" serializer="javaSerializer" deserializer="javaDeserializer"
    so-keep-alive="true" />

<int:channel id="input" />

<int:channel id="replies">
    <int:queue />
</int:channel>
<int-ip:tcp-outbound-channel-adapter
    id="outboundClient" channel="input" connection-factory="client" />

<int-ip:tcp-inbound-channel-adapter
    id="inboundClient" channel="replies" connection-factory="client"
    client-mode="true" auto-startup="true" />

我的 Tcp 服务器：

while(true)
  {
     try
     {
        System.out.println("Waiting for client on port " +
        serverSocket.getLocalPort() + "...");

        Socket server = serverSocket.accept();
        System.out.println("Just connected to "
              + server.getRemoteSocketAddress());

        DataOutputStream out =
             new DataOutputStream(server.getOutputStream());
        out.write("ACK\r\n".getBytes());

        out.flush();

       //server.close();

     }catch(SocketTimeoutException s)
     {
        System.out.println("Socket timed out!");
        break;
     }catch(IOException e)
     {
        e.printStackTrace();
        break;
     } 
  }

服务器日志：

正在端口 5678 上等待客户端... 刚刚连接到/127.0.0.1:56108 正在端口 5678 上等待客户端...

我的 TcpClient：

    final GenericXmlApplicationContext context = new GenericXmlApplicationContext();

    context.load("classpath:config.xml");

    context.registerShutdownHook();
    context.refresh();

    final SimpleGateway gateway = context.getBean(SimpleGateway.class);
    int i=0;

    while(i++<10){

    String h = gateway.receive();
    System.out.println("Received message "+h);

    }try {
        Thread.sleep(1000);
    } catch (InterruptedException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    }

我的客户日志：

收到消息确认

收到的消息收到消息收到消息收到消息收到消息收到消息收到消息收到消息收到消息

我的自定义反序列化器：

@Override
public String deserialize(InputStream inputStream) throws IOException {
    // TODO Auto-generated method stub
    StringBuilder builder = new StringBuilder();


    int size = inputStream.available();

    int c;
    for (int i = 0; i < size; ++i) {
        c = inputStream.read();

        if(c!=-1){
        builder.append((char)c);
        }
        else{
            break;
        }
    }


    return builder.toString();
}

我的网关：

public interface SimpleGateway {

    public String receive();

}

如果您还有其他问题，请告诉我。

【问题讨论】：

标签： spring-integration

【解决方案1】：

您的服务器只发送一条消息，然后接受新连接。

编辑

我在收到消息的那一刻触发某些功能？

我不太明白问题的那一部分 - 您当前将 input 通道连接到出站通道适配器，该适配器只是将其重新写回。

你可以做这样的事情......

<int:object-to-string-transformer input-channel="input" output-channel="next" />

<int:service-activator input-channel="next" method="foo">
    <bean class='foo.Foo" />
</int:service-activator>

public class Foo {

    public void foo(String payload) {
        ...
    }

}

如果你想处理byte[]，你可以省略转换器并使用foo(byte[] bytes)。

【讨论】：

是的！太明显了。无法指出它。谢谢加里。问题的另一部分是，是否可以通知我的客户有关写入套接字的数据，以便我可以将我的功能放在那里，以处理我的消息。
好的，我从您的编辑中了解到，我将消息放入`foo()`。但我希望，当数据写入套接字时，应该通知客户端。换句话说，我不希望我的客户端轮询传入的数据，而是在数据写入套接字时得到通知。
不是轮询，是“消息驱动”——服务器发送一些数据时，作为消息发送（反序列化后）。

【解决方案2】：

我和你有类似的问题。我花了一些时间检查 spring 源代码。这是我发现的。

这就是春天所做的。

第一个线程。当你调用网关的任何方法时，Spring 会调用 TcpOutboundGateway.handleRequestMessage，它会创建 TcpAysReply 并保存到 TcpOutboundGateway.pendingReplies。该方法会将其挂在那里（取决于 TcpAysReply）等待回复。

  Message<?> getReply() {
      try {
          if (!this.latch.await(this.remoteTimeout, TimeUnit.MILLISECONDS)) {
              return null;
          }
      }
      catch (@SuppressWarnings("unused") InterruptedException e) {
          Thread.currentThread().interrupt();
      }
      boolean waitForMessageAfterError = true;
      while (this.reply instanceof ErrorMessage) {
          if (waitForMessageAfterError) {
              /*
               * Possible race condition with NIO; we might have received the close
               * before the reply, on a different thread.
               */
              logger.debug("second chance");
              try {
                  this.secondChanceLatch
                          .await(TcpOutboundGateway.this.secondChanceDelay, TimeUnit.SECONDS); // NOSONAR
              }
              catch (@SuppressWarnings("unused") InterruptedException e) {
                  Thread.currentThread().interrupt();
                  doThrowErrorMessagePayload();
              }
              waitForMessageAfterError = false;
          }
          else {
              doThrowErrorMessagePayload();
          }
      }
      return this.reply;
  }

第二个线程。 Spring 将运行 TcpNetConnection 的 run() 方法

  while (true) {
      if (!receiveAndProcessMessage()) {
          break;
      }
  }

这个方法最终调用TcpOutboundGatway.onMessage()

   // some code here    
    TcpOutboundGateway.AsyncReply reply = this.pendingReplies.get(connectionId);
    // some doe here

如果reply != null 则释放第一个线程，然后网关方法返回数据。

所以有两种情况会发生：

正常情况，第一个线程创建AsyncReply，第二个线程释放AsyncReply。在这种情况下，您可以正常从网关方法中获取消息。
异常情况，在创建 AysnReply 之前，第二个线程运行释放 AsyncReply（您的情况）。在这种情况下，您无法从 Gateway 方法取回数据。

您仍然收到第一条消息的原因是您第一次调用网关时，Spring 尚未运行第二个线程。这就是为什么第一次调用是正常情况的原因。

解决方案：在TcpOutboundGatway.onMessage()

 if (reply == null) {
            if (message instanceof ErrorMessage) {
                /*
                 * Socket errors are sent here so they can be conveyed to any waiting thread.
                 * If there's not one, simply ignore.
                 */
                return false;
            }
            else {
                if (unsolicitedSupported(message)) {
                    return false;
                }
                String errorMessage = "Cannot correlate response - no pending reply for " + connectionId;
                logger.error(errorMessage);
                publishNoConnectionEvent(message, connectionId, errorMessage);
                return false;
            }
        }

如果 Spring 找不到回复，它会将消息发送到未经请求的频道。您只需要简单地定义未经请求的渠道和 ServiceActivator。例如

@Bean
@ServiceActivator(inputChannel = "toTcp")
public MessageHandler tcpOutGate(AbstractClientConnectionFactory connectionFactory) {
    TcpOutboundGateway gate = new TcpOutboundGateway();
    gate.setRemoteTimeout(100000000);
    gate.setUnsolicitedMessageChannelName("unSolicited");
    gate.setConnectionFactory(connectionFactory);
    return gate;
}

@Bean
@ServiceActivator(inputChannel = "unSolicited")
public MessageHandler resolveFirstMessage() {
    return message -> {
        System.out.println("Received message "+message);
    };
}

【讨论】：