使用 Netty4 和 Apache Camel 的全双工 TCP 连接答案

【问题标题】：Full duplex TCP connections using Netty4 and Apache Camel使用 Netty4 和 Apache Camel 的全双工 TCP 连接
【发布时间】：2019-07-19 17:15:31
【问题描述】：

对于我正在从事的物联网项目，我正在研究我们的“套接字处理程序”的下一个增强版本，它已经超过 5 年了，除了处理与物联网设备的套接字连接外，它已经发展成为一个巨大的野兽在线程处理中，管理起来真的很痛苦。

对于我的全部重写，我正在研究 Apache Camel 作为路由和转换工具包，并了解它如何帮助我们将处理步骤拆分为微服务，通过消息队列松散耦合。

然而，我难以理解的一件事是如何实现以下逻辑“Apache Camel 方式”：物联网设备发送一条初始消息，其中包含其 id、一些额外的标头和消息有效负载。除了提取消息负载并将其路由到通道之外，我还需要使用设备 ID 来检查以设备 ID 命名的消息队列，以查找必须通过接收到的同一套接字连接发送到设备的任何命令初始消息。

虽然看起来Camel中包含的Netty4可以处理同步双工通信，但我看不出上面的逻辑如何在Camel Netty4组件中实现。骆驼路由似乎只是一种方式。

有没有正确的方法可以做到这一点，还是我应该忘记使用骆驼来完成这项工作而只使用 Netty4 裸机？

【问题讨论】：

标签： apache-camel netty

【解决方案1】：

使用两条 Camel 路由可以重现全双工通信：

感谢使用reuseChannel 属性
由于这种全双工通信将通过两条不同的 Camel 路由实现，因此必须将 sync 属性设置为 false。

这里是第一条路线：

from("netty4:tcp://{{tcpAddress}}:{{tcpPort}}?decoders=#length-decoder,#string-decoder&encoders=#length-encoder,#bytearray-encoder&sync=false&reuseChannel=true") .bean("myMessageService", "receiveFromTCP").to("jms:queue:<name>")

这第一个路由将创建一个 TCP/IP 消费者，这要归功于服务器套接字（由于属性 clientMode，也可以使用客户端套接字）

由于我们想要重用刚刚创建的连接，因此初始化解码器和编码器非常重要，这要归功于 bean（请参阅更多信息）。这个 bean 将负责使用在第一个路由中创建的Channel 发送数据（Netty Channel 包含一个管道，用于在从 TCP/IP 接收/发送到 TCP/IP 之前解码/编码消息。

现在，我们想要将一些数据发送回连接到第一条路由的消费者（来自）端点的部分。由于经典的生产者端点（to），我们无法做到这一点，我们使用一个 bean 对象：

from("jsm:queue:<name>").bean("myMessageService", "sendToTCP");

这里是 bean 代码：

public class MessageService {
    
  private Channel openedChannel;

     public void sendToTCP(final Exchange exchange) {
      // opened channel will use encoders before writing on the socket already 
      // created in the first route
      openedChannel.writeAndFlush(exchange.getIn().getBody());
     }

     public void receiveFromTCP(final Exchange exchange) {
      // record the channel created in the first route.
      this.openedChannel = exchange.getProperty(NettyConstants.NETTY_CHANNEL, Channel.class);
     }
}

当然，

两个路由使用相同的 bean 实例，您需要使用注册表来执行此操作：

SimpleRegistry simpleRegistry = new SimpleRegistry();
simpleRegistry.put("myMessageService", new MessageService());

由于 bean 用于两个不同的异步路由，您将不得不处理一些未执行的情况，例如保护 openChannel 成员的访问，或处理意外断开连接。

这篇文章帮助我找到了这个解决方案： how-to-send-a-response-back-over-a-established-tcp-connection-in-async-mode-usin

reuseChannel property documentation

【讨论】：

【解决方案2】：

camel route 结束后，exchange 的 body 和 headers 将作为响应返回给请求者。

【讨论】：