TCP 和 UDP 支持
TCP 和 UDP 支持
Spring Integration 提供了用于通过互联网协议接收和发送消息的通道适配器。 提供 UDP（用户数据报协议）和 TCP（传输控制协议）适配器。 每个适配器都通过底层协议提供单向通信。 此外，Spring Integration 还提供了简单的入站和出站 TCP 网关。 当需要双向通信时使用这些。
您需要将此依赖项包含在您的项目中：
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-ip</artifactId>
<version>6.1.9</version>
</dependency>compile "org.springframework.integration:spring-integration-ip:6.1.9"
介绍
提供 UDP 入站和出站通道适配器各两种类型：
-
UnicastSendingMessageHandler将数据报数据包发送到单个目标。 -
UnicastReceivingChannelAdapter接收传入的数据报数据包。 -
MulticastSendingMessageHandler将数据报数据包发送（广播）到组播地址。 -
MulticastReceivingChannelAdapter通过加入组播地址来接收传入的数据报数据包。
提供 TCP 入站和出站通道适配器：
-
TcpSendingMessageHandler通过 TCP 发送消息。 -
TcpReceivingChannelAdapter通过 TCP 接收消息。
提供入站 TCP 网关。 它允许简单的请求-响应处理。 虽然网关可以支持任意数量的连接，但每个连接只能串行处理。 从套接字读取的线程在再次读取之前等待并发送响应。 如果连接工厂配置为一次性连接，则在套接字超时后关闭连接。
提供出站TCP网关。 它允许简单的请求-响应处理。 如果为一次性连接配置了关联的连接工厂，那么将立即为每个新请求创建一个新连接。 否则，如果连接正在使用中，则调用线程会阻塞连接，直到收到响应或发生超时或 I/O 错误。
TCP 和 UDP 入站通道适配器以及 TCP 入站网关支持error-channel属性。
这提供了与输入GatewayProxyFactoryBean.
UDP 适配器
本节介绍如何配置和使用 UDP 适配器。
出站 UDP 适配器（XML 配置）
以下示例配置 UDP 出站通道适配器：
<int-ip:udp-outbound-channel-adapter id="udpOut"
host="somehost"
port="11111"
multicast="false"
socket-customizer="udpCustomizer"
channel="exampleChannel"/>设置multicast自true，您还应该在主机属性中提供组播地址。 |
UDP 是一种高效但不可靠的协议。
Spring Integration 添加了两个属性以提高可靠性：check-length和acknowledge.
什么时候check-length设置为true，适配器在消息数据前面有一个长度字段（按网络字节顺序排列为四个字节）。
这使接收方能够验证接收到的数据包的长度。
如果接收系统使用的缓冲区太短而无法包含数据包，则可以截断数据包。
这lengthheader 提供了一种检测此问题的机制。
从 4.3 版本开始，您可以将port自0，在这种情况下，作系统选择端口。
可以通过调用getPort()适配器启动后isListening()返回true.
从 5.3.3 版本开始，您可以添加SocketCustomizerbean 来修改DatagramSocket创建后（例如，调用setTrafficClass(0x10)).
以下示例显示了一个出站通道适配器，该适配器向数据报数据包添加了长度检查：
<int-ip:udp-outbound-channel-adapter id="udpOut"
host="somehost"
port="11111"
multicast="false"
check-length="true"
channel="exampleChannel"/>数据包的接收方还必须配置为预期实际数据之前的长度。
对于 Spring Integration UDP 入站通道适配器，请将其check-length属性。 |
第二个可靠性改进允许使用应用程序级确认协议。 接收方必须在指定时间内向发送方发送确认。
以下示例显示了一个出站通道适配器，该适配器向数据报数据包添加长度检查并等待确认：
<int-ip:udp-outbound-channel-adapter id="udpOut"
host="somehost"
port="11111"
multicast="false"
check-length="true"
acknowledge="true"
ack-host="thishost"
ack-port="22222"
ack-timeout="10000"
channel="exampleChannel"/>设置acknowledge自true意味着数据包的接收者可以解释添加到包含确认数据（主机和端口）的数据包中的标头。
最有可能的是，接收者是 Spring Integration 入站通道适配器。 |
当多播为 true 时，附加属性 （min-acks-for-success） 指定必须在ack-timeout. |
从 4.3 版本开始，您可以将ackPort自0，在这种情况下，作系统选择端口。
出站 UDP 适配器（Java 配置）
以下示例显示如何使用 Java 配置出站 UDP 适配器：
@Bean
@ServiceActivator(inputChannel = "udpOut")
public UnicastSendingMessageHandler handler() {
return new UnicastSendingMessageHandler("localhost", 11111);
}
（或MulticastSendingChannelAdapter用于组播）。
出站 UDP 适配器（Java DSL 配置）
以下示例显示如何使用 Java DSL 配置出站 UDP 适配器：
@Bean
public IntegrationFlow udpOutFlow() {
return f -> f.handle(Udp.outboundAdapter("localhost", 1234)
.configureSocket(socket -> socket.setTrafficClass(0x10)))
.get();
}
入站 UDP 适配器（XML 配置）
以下示例演示如何配置基本单播入站 udp 通道适配器。
<int-ip:udp-inbound-channel-adapter id="udpReceiver"
channel="udpOutChannel"
port="11111"
receive-buffer-size="500"
multicast="false"
socket-customizer="udpCustomizer"
check-length="true"/>以下示例演示如何配置基本组播入站 udp 通道适配器：
<int-ip:udp-inbound-channel-adapter id="udpReceiver"
channel="udpOutChannel"
port="11111"
receive-buffer-size="500"
multicast="true"
multicast-address="225.6.7.8"
check-length="true"/>默认情况下，不对入站数据包执行反向 DNS 查找：在未配置 DNS 的环境（例如 Docker 容器）中，这可能会导致连接延迟。
要将 IP 地址转换为主机名以用于邮件头，可以通过将lookup-host属性设置为true.
从 5.3.3 版本开始，您可以添加SocketCustomizerbean 来修改DatagramSocket创建后。
它用于接收套接字和为发送 ack 创建的任何套接字。
入站 UDP 适配器（Java 配置）
以下示例显示如何使用 Java 配置入站 UDP 适配器：
@Bean
public UnicastReceivingChannelAdapter udpIn() {
UnicastReceivingChannelAdapter adapter = new UnicastReceivingChannelAdapter(11111);
adapter.setOutputChannelName("udpChannel");
return adapter;
}
以下示例显示如何使用 Java DSL 配置入站 UDP 适配器：
入站 UDP 适配器（Java DSL 配置）
@Bean
public IntegrationFlow udpIn() {
return IntegrationFlow.from(Udp.inboundAdapter(11111))
.channel("udpChannel")
.get();
}
服务器监听事件
从 5.0.2 版开始，UdpServerListeningEvent在入站适配器启动并已开始侦听时发出。
当适配器配置为侦听端口时，这很有用0，这意味着作系统选择端口。
它也可以代替轮询isListening()，如果您需要等待才能开始将连接到套接字的其他进程。
高级出站配置
这<int-ip:udp-outbound-channel-adapter> (UnicastSendingMessageHandler） 有destination-expression和socket-expression选项。
您可以使用destination-expression作为硬编码的运行时替代方案host-portpair 来确定传出数据报数据包的目标地址，以根据requestMessage（使用求值上下文的根对象）。
表达式必须计算为URI一个String在 URI 样式中（参见 RFC-2396），或SocketAddress.
您还可以使用 inboundIpHeaders.PACKET_ADDRESS标头。
在框架中，DatagramPacketMessageMapper当我们在UnicastReceivingChannelAdapter并将它们转换为消息。
标头值正是DatagramPacket.getSocketAddress()传入数据报。
使用socket-expression，则出站通道适配器可以使用（例如）入站通道适配器套接字通过接收数据报的同一端口发送数据报。
在我们的应用程序用作 UDP 服务器并且客户端在网络地址转换 （NAT） 后面运行的情况下，它非常有用。
此表达式必须计算为DatagramSocket.
这requestMessage用作评估上下文的根对象。
您不能使用socket-expression参数替换为multicast和acknowledge参数。
以下示例显示如何使用转换为大写并使用套接字的转换器配置 UDP 入站通道适配器：
<int-ip:udp-inbound-channel-adapter id="inbound" port="0" channel="in" />
<int:channel id="in" />
<int:transformer expression="new String(payload).toUpperCase()"
input-channel="in" output-channel="out"/>
<int:channel id="out" />
<int-ip:udp-outbound-channel-adapter id="outbound"
socket-expression="@inbound.socket"
destination-expression="headers['ip_packetAddress']"
channel="out" />以下示例显示了 Java DSL 的等效配置：
@Bean
public IntegrationFlow udpEchoUpcaseServer() {
return IntegrationFlow.from(Udp.inboundAdapter(11111).id("udpIn"))
.<byte[], String>transform(p -> new String(p).toUpperCase())
.handle(Udp.outboundAdapter("headers['ip_packetAddress']")
.socketExpression("@udpIn.socket"))
.get();
}
TCP 连接工厂
概述
对于 TCP，基础连接的配置是通过使用连接工厂提供的。 提供两种类型的连接工厂：客户端连接工厂和服务器连接工厂。 客户端连接工厂建立传出连接。 服务器连接工厂侦听传入连接。
出站通道适配器使用客户机连接工厂，但您也可以提供对入站通道适配器的客户机连接工厂的引用。 该适配器接收在出站适配器创建的连接上接收的任何传入消息。
入站通道适配器或网关使用服务器连接工厂。 （事实上，没有一个，连接工厂就无法运行）。 您还可以提供对出站适配器的服务器连接工厂的引用。 然后，您可以使用该适配器向同一连接上的传入邮件发送回复。
仅当回复包含ip_connectionId由连接工厂插入到原始消息中的标头。 |
| 这是在入站和出站适配器之间共享连接工厂时执行的消息关联范围。 这种共享允许通过 TCP 进行异步双向通信。 默认情况下，仅使用 TCP 传输有效负载信息。 因此，任何消息关联都必须由下游组件（例如聚合器或其他终结点）执行。 3.0 版中引入了对传输选定标头的支持。 有关详细信息，请参阅 TCP 消息关联。 |
您可以向每种类型的最多一个适配器提供对连接工厂的引用。
Spring Integration提供了使用该连接工厂的连接工厂java.net.Socket和java.nio.channel.SocketChannel.
以下示例显示了一个简单的服务器连接工厂，该工厂使用java.net.Socket连接：
<int-ip:tcp-connection-factory id="server"
type="server"
port="1234"/>以下示例显示了一个简单的服务器连接工厂，该工厂使用java.nio.channel.SocketChannel连接：
<int-ip:tcp-connection-factory id="server"
type="server"
port="1234"
using-nio="true"/>从 Spring Integration 版本 4.2 开始，如果服务器配置为侦听随机端口（通过将端口设置为0），您可以使用以下命令获取作系统选择的实际端口getPort().
也getServerSocketAddress()让您获得完整的SocketAddress.
请参阅Javadoc 的TcpServerConnectionFactory接口了解更多信息。 |
<int-ip:tcp-connection-factory id="client"
type="client"
host="localhost"
port="1234"
single-use="true"
so-timeout="10000"/>以下示例显示了使用java.net.Socketconnections 并为每条消息创建一个新连接：
<int-ip:tcp-connection-factory id="client"
type="client"
host="localhost"
port="1234"
single-use="true"
so-timeout="10000"
using-nio=true/>从 V5.2 开始，客户端连接工厂支持connectTimeout，以秒为单位指定，默认为 60。
消息分界（序列化程序和反序列化程序）
TCP 是一种流协议。 这意味着必须为通过 TCP 传输的数据提供一些结构，以便接收方可以将数据划分为离散的消息。 连接工厂配置为使用序列化程序和反序列化程序在消息有效负载和通过 TCP 发送的位之间进行转换。 这是通过分别为入站和出站消息提供反序列化程序和序列化程序来实现的。 Spring Integration 提供了许多标准序列化器和反序列化器。
ByteArrayCrlfSerializer*将字节数组转换为字节流，后跟回车符和换行符 （\r\n). 这是默认的序列化程序（和反序列化程序），可以（例如）与 telnet 一起使用作为客户端。
这ByteArraySingleTerminatorSerializer*将字节数组转换为字节流，后跟单个终止字符（默认值为0x00).
这ByteArrayLfSerializer*将字节数组转换为字节流，后跟单个换行字符 （0x0a).
这ByteArrayStxEtxSerializer*将字节数组转换为前面有 STX （0x02），然后是 ETX （0x03).
这ByteArrayLengthHeaderSerializer将字节数组转换为前面按网络字节顺序（大端）排列的二进制长度的字节流。
这是一个高效的反序列化器，因为它不必解析每个字节来查找终止字符序列。
它还可用于包含二进制数据的有效负载。
前面的序列化程序仅支持有效负载中的文本。
长度标头的默认大小为 4 个字节（一个整数），允许最多 （2^31 - 1） 字节的消息。
但是，length对于最多 255 字节的消息，标头可以是单个字节（无符号），对于最多 （2^16 - 1） 字节的消息，标头可以是无符号短字节（2 字节）。
如果您需要任何其他格式的标头，您可以子类化ByteArrayLengthHeaderSerializer并为readHeader和writeHeader方法。
绝对最大数据大小为 （2^31 - 1） 字节。
从版本 5.2 开始，标头值除了有效负载之外还可以包括标头的长度。
将inclusive属性来启用该机制（必须为生产者和使用者设置相同的机制）。
这ByteArrayRawSerializer*，将字节数组转换为字节流，并且不添加额外的消息分界数据。
使用此序列化程序 （和反序列化程序） ，客户端以有序方式关闭套接字来指示消息的结束。
使用此序列化程序时，消息接收将挂起，直到客户端关闭套接字或发生超时。
超时不会导致消息。
当使用此序列化程序并且客户端是 Spring Integration 应用程序时，客户端必须使用配置为single-use="true".
这样做会导致适配器在发送消息后关闭套接字。
序列化程序本身不会关闭连接。
您应该仅将此序列化程序用于通道适配器（而不是网关）使用的连接工厂，并且连接工厂应由入站或出站适配器使用，但不能同时由两者使用。
也可以看看ByteArrayElasticRawDeserializer，在本节后面。
但是，从版本 5.2 开始，出站网关有一个新属性closeStreamAfterSend;这允许使用原始序列化程序/反序列化程序，因为 EOF 会向服务器发出信号，同时保持连接打开以接收回复。
在 4.2.2 版之前，当使用非阻塞 I/O （NIO） 时，该序列化器将超时（读取期间）视为文件结束，并且到目前为止读取的数据作为消息发出。
这是不可靠的，不应用于分隔消息。
它现在将此类情况视为例外。
万一以这种方式使用它，您可以通过将treatTimeoutAsEndOfMessageconstructor 参数设置为true. |
其中每一个都是AbstractByteArraySerializer，它实现了org.springframework.core.serializer.Serializer和org.springframework.core.serializer.Deserializer.
为了向后兼容，使用AbstractByteArraySerializer对于序列化，也接受String首先转换为字节数组。
这些序列化程序和反序列化程序中的每一个都将包含相应格式的输入流转换为字节数组有效负载。
为了避免由于行为不良的客户端（不遵守已配置序列化程序的协议）而导致内存耗尽，这些序列化程序会强制设置最大消息大小。
如果传入邮件超过此大小，则会引发异常。
默认最大消息大小为 2048 字节。
您可以通过设置maxMessageSize财产。
如果您使用缺省序列化程序或反序列化程序并希望增加最大消息大小，则必须将最大消息大小声明为显式 Bean，并使用maxMessageSize属性设置并配置连接工厂以使用该 Bean。
本节前面标记的类使用中间缓冲区，并将解码的数据复制到大小正确的最终缓冲区。
从 4.3 版开始，您可以通过设置*poolSize属性，让这些原始缓冲区被重用，而不是为每条消息分配和丢弃，这是默认行为。
将属性设置为负值将创建一个没有边界的池。
如果池是有界的，您还可以将poolWaitTimeout属性（以毫秒为单位），之后如果没有可用的缓冲区，则会引发异常。
它默认为无大。
此类异常会导致套接字关闭。
如果您希望在自定义反序列化器中使用相同的机制，您可以扩展AbstractPooledBufferByteArraySerializer（而不是它的超类，AbstractByteArraySerializer）并实现doDeserialize()而不是deserialize().
缓冲区会自动返回到池中。AbstractPooledBufferByteArraySerializer还提供了一种方便的实用方法：copyToSizedArray().
5.0 版本添加了ByteArrayElasticRawDeserializer.
这类似于ByteArrayRawSerializer上面，除了没有必要设置一个maxMessageSize.
在内部，它使用ByteArrayOutputStream这让缓冲液根据需要增长。
客户端必须有序关闭套接字，以发出消息结束的信号。
| 仅当对等方受信任时，才应使用此反序列化程序;由于内存不足，它容易受到 DoS 附加的影响。 |
这MapJsonSerializer使用JacksonObjectMapper在Map和 JSON。
您可以将此序列化程序与MessageConvertingTcpMessageMapper和MapMessageConverter以 JSON 格式传输选定的标头和有效负载。
JacksonObjectMapper无法划分流中的消息。
因此，MapJsonSerializer需要委托给另一个序列化程序或反序列化程序来处理消息分界。
默认情况下，一个ByteArrayLfSerializer，则生成格式为<json><LF>在网络上，但您可以将其配置为使用其他。
（下一个示例演示了如何执行此作。 |
最终的标准序列化程序是org.springframework.core.serializer.DefaultSerializer，您可以使用它来转换具有 Java 序列化的可序列化对象。org.springframework.core.serializer.DefaultDeserializer用于对包含可序列化对象的流进行入站反序列化。
如果您不想使用默认序列化器和反序列化程序 （ByteArrayCrLfSerializer），您必须将serializer和deserializer属性。
以下示例显示了如何执行此作：
<bean id="javaSerializer"
class="org.springframework.core.serializer.DefaultSerializer" />
<bean id="javaDeserializer"
class="org.springframework.core.serializer.DefaultDeserializer" />
<int-ip:tcp-connection-factory id="server"
type="server"
port="1234"
deserializer="javaDeserializer"
serializer="javaSerializer"/>使用java.net.Socket连接，并在网络上使用 Java 序列化。
有关连接工厂上可用属性的完整详细信息，请参阅本节末尾的参考。
默认情况下，不对入站数据包执行反向 DNS 查找：在未配置 DNS 的环境（例如 Docker 容器）中，这可能会导致连接延迟。
要将 IP 地址转换为主机名以用于邮件头，可以通过将lookup-host属性设置为true.
| 您还可以修改套接字和套接字工厂的属性。 有关更多信息，请参阅 SSL/TLS 支持。 如前所述，如果使用 SSL 或不使用 SSL，则可以进行此类修改。 |
自定义序列化器和解串化器
如果你的数据不是标准反序列化程序之一支持的格式，你可以实现自己的格式;还可以实现自定义序列化程序。
要实现自定义序列化器和反序列化器对，请实现org.springframework.core.serializer.Deserializer和org.springframework.core.serializer.Serializer接口。
当反序列化程序检测到消息之间的关闭输入流时，它必须引发SoftEndOfStreamException;这是向框架发出的信号，表明收盘价是“正常”的。
如果在解码消息时关闭流，则应引发其他异常。
从 5.2 版本开始，SoftEndOfStreamException现在是一个RuntimeException而不是扩展IOException.
TCP 缓存客户端连接工厂
如前所述，TCP 套接字可以是“一次性”（一个请求或响应）或共享的。 在大容量环境中，共享套接字在出站网关中表现不佳，因为套接字一次只能处理一个请求或响应。
为了提高性能，您可以使用协作通道适配器而不是网关，但这需要应用程序级消息关联。 有关详细信息，请参阅 TCP 消息关联。
Spring Integration 2.2 引入了一个缓存客户端连接工厂，它使用共享套接字池，允许网关使用共享连接池处理多个并发请求。
TCP 故障转移客户端连接工厂
您可以配置支持故障转移到一个或多个其他服务器的 TCP 连接工厂。 发送消息时，工厂会遍历其所有配置的工厂，直到可以发送消息或找不到连接。 最初，使用已配置列表中的第一个工厂。 如果连接随后失败，则下一个工厂将成为当前工厂。 以下示例演示如何配置故障转移客户端连接工厂：
<bean id="failCF" class="o.s.i.ip.tcp.connection.FailoverClientConnectionFactory">
<constructor-arg>
<list>
<ref bean="clientFactory1"/>
<ref bean="clientFactory2"/>
</list>
</constructor-arg>
</bean>使用故障转移连接工厂时，singleUse属性必须在工厂本身和它配置为使用的工厂列表之间保持一致。 |
当与共享连接 （singleUse=false):
-
refreshSharedInterval -
closeOnRefresh
根据上述配置，请考虑以下方案：
比方说clientFactory1无法建立连接，但clientFactory2能。
当failCF getConnection()方法在refreshSharedInterval已通过，我们将再次尝试使用clientFactory1;如果成功，则连接到clientFactory2将关闭。
如果closeOnRefresh是false，“旧”连接将保持打开状态，如果第一个工厂再次出现故障，将来可能会重复使用。
设置refreshSharedInterval仅在该时间到期后才尝试重新连接第一家工厂;将其设置为Long.MAX_VALUE（默认）如果您只想在当前连接失败时故障恢复到第一个工厂。
设置closeOnRefresh在刷新后关闭“旧”连接实际上会创建一个新连接。
如果任何委托工厂是CachingClientConnectionFactory因为连接缓存在那里处理;在这种情况下，将始终查阅连接工厂列表以获取连接。 |
从 5.3 版开始，这些默认为Long.MAX_VALUE和true因此，工厂仅在当前连接失败时才尝试故障回复。
要恢复到以前版本的默认行为，请将它们设置为0和false.
另请参阅测试连接。
TCP 线程关联连接工厂
Spring Integration 5.0 版引入了这个连接工厂。
它将连接绑定到调用线程，并且每次该线程发送消息时都会重复使用相同的连接。
这种情况一直持续到连接关闭（由服务器或网络）或直到线程调用releaseConnection()方法。
连接本身由另一个客户端工厂实现提供，必须将其配置为提供非共享（一次性）连接，以便每个线程获得连接。
以下示例显示如何配置 TCP 线程关联连接工厂：
@Bean
public TcpNetClientConnectionFactory cf() {
TcpNetClientConnectionFactory cf = new TcpNetClientConnectionFactory("localhost",
Integer.parseInt(System.getProperty(PORT)));
cf.setSingleUse(true);
return cf;
}
@Bean
public ThreadAffinityClientConnectionFactory tacf() {
return new ThreadAffinityClientConnectionFactory(cf());
}
@Bean
@ServiceActivator(inputChannel = "out")
public TcpOutboundGateway outGate() {
TcpOutboundGateway outGate = new TcpOutboundGateway();
outGate.setConnectionFactory(tacf());
outGate.setReplyChannelName("toString");
return outGate;
}
测试连接
在某些情况下，在首次打开连接时发送某种运行状况检查请求可能很有用。 其中一种情况可能是在使用 TCP 故障转移客户端连接工厂时，以便在所选服务器允许打开连接但报告连接不正常时进行故障转移。
为了支持此功能，请添加一个connectionTest到客户端连接工厂。
/**
* Set a {@link Predicate} that will be invoked to test a new connection; return true
* to accept the connection, false the reject.
* @param connectionTest the predicate.
* @since 5.3
*/
public void setConnectionTest(@Nullable Predicate<TcpConnectionSupport> connectionTest) {
this.connectionTest = connectionTest;
}
要测试连接，请将临时侦听器附加到测试中的连接。 如果测试失败，则连接将关闭并引发异常。 当与 TCP 故障转移客户端连接工厂一起使用时，这会触发尝试下一台服务器。
| 只有来自服务器的第一个回复才会发送给测试侦听器。 |
在以下示例中，如果服务器回复，则服务器被视为正常PONG当我们发送PING.
Message<String> ping = new GenericMessage<>("PING");
byte[] pong = "PONG".getBytes();
clientFactory.setConnectionTest(conn -> {
CountDownLatch latch = new CountDownLatch(1);
AtomicBoolean result = new AtomicBoolean();
conn.registerTestListener(msg -> {
if (Arrays.equals(pong, (byte[]) msg.getPayload())) {
result.set(true);
}
latch.countDown();
return false;
});
conn.send(ping);
try {
latch.await(10, TimeUnit.SECONDS);
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
return result.get();
});
TCP 连接拦截器
您可以使用对TcpConnectionInterceptorFactoryChain.
可以使用拦截器向连接添加行为，例如协商、安全性和其他选项。
该框架当前没有提供拦截器，但请参阅InterceptedSharedConnectionTests在源存储库中举个例子。
这HelloWorldInterceptor用于测试用例的工作原理如下：
拦截器首先配置了客户端连接工厂。 当第一条消息通过拦截的连接发送时，拦截器通过连接发送“Hello”，并期望收到“world！”。 发生这种情况时，协商完成并发送原始消息。 此外，使用相同连接的消息无需任何额外协商即可发送。
当配置了服务器连接工厂时，拦截器要求第一条消息是“Hello”，如果是，则返回“world！”。 否则，它会抛出导致连接关闭的异常。
都TcpConnection方法被拦截。
拦截器实例由拦截器工厂为每个连接创建。
如果拦截器是有状态的，则工厂应为每个连接创建一个新实例。
如果没有状态，则同一个拦截器可以包装每个连接。
拦截器工厂被添加到拦截器工厂链的配置中，您可以通过将interceptor-factory属性。
拦截器必须扩展TcpConnectionInterceptorSupport.
工厂必须实现TcpConnectionInterceptorFactory接口。TcpConnectionInterceptorSupport具有直通方法。
通过扩展此类，您只需要实现您希望拦截的方法。
以下示例演示如何配置连接拦截器工厂链：
<bean id="helloWorldInterceptorFactory"
class="o.s.i.ip.tcp.connection.TcpConnectionInterceptorFactoryChain">
<property name="interceptors">
<array>
<bean class="o.s.i.ip.tcp.connection.HelloWorldInterceptorFactory"/>
</array>
</property>
</bean>
<int-ip:tcp-connection-factory id="server"
type="server"
port="12345"
using-nio="true"
single-use="true"
interceptor-factory-chain="helloWorldInterceptorFactory"/>
<int-ip:tcp-connection-factory id="client"
type="client"
host="localhost"
port="12345"
single-use="true"
so-timeout="100000"
using-nio="true"
interceptor-factory-chain="helloWorldInterceptorFactory"/>TCP 连接事件
从 3.0 版开始，更改为TcpConnection实例由TcpConnectionEvent实例。TcpConnectionEvent是ApplicationEvent因此，任何人都可以接收ApplicationListener在ApplicationContext— 例如，事件入站通道适配器。
TcpConnectionEvents具有以下属性：
-
connectionId：连接标识符，您可以在消息标头中使用该标识符将数据发送到连接。 -
connectionFactoryName：连接所属的连接工厂的 Bean 名称。 -
throwable：这Throwable（对于TcpConnectionExceptionEvent仅事件）。 -
source：这TcpConnection. 例如，您可以使用它来确定远程 IP 地址getHostAddress()（需要演员）。
此外，从 4.0 版开始，TCP 连接工厂中讨论的标准反序列化程序现在发出TcpDeserializationExceptionEvent在解码数据流时遇到问题的情况。
这些事件包含异常、正在生成的缓冲区以及异常发生点的缓冲区 （（如果可用） 的偏移量。
应用程序可以使用普通的ApplicationListener或ApplicationEventListeningMessageProducer（参见接收 Spring Application 事件）来捕获这些事件，从而分析问题。
从 4.0.7 和 4.1.3 版本开始，TcpConnectionServerExceptionEvent每当服务器套接字上发生意外异常时，都会发布实例（例如BindException当服务器套接字正在使用时）。
这些事件引用了连接工厂和原因。
从 4.2 版本开始，TcpConnectionFailedCorrelationEvent每当端点（入站网关或协作出站通道适配器）收到无法路由到连接的消息时，就会发布实例，因为ip_connectionIdheader 无效。
当收到延迟回复（发送方线程超时）时，出站网关也会发布此事件。
该事件包含连接 ID 以及cause属性，其中包含失败的消息。
从 4.3 版开始，TcpConnectionServerListeningEvent在启动服务器连接工厂时发出。
当工厂配置为侦听端口时，这很有用0，这意味着作系统选择端口。
它也可以代替轮询isListening()，如果您需要等待才能启动连接到套接字的其他进程。
| 为了避免延迟侦听线程接受连接，事件将发布在单独的线程上。 |
从 4.3.2 版开始，TcpConnectionFailedEvent每当无法创建客户端连接时发出。
事件的源是连接工厂，您可以使用它来确定无法建立连接的主机和端口。
TCP 适配器
提供了使用前面提到的连接工厂的 TCP 入站和出站通道适配器。
这些适配器有两个相关属性：connection-factory和channel.
这connection-factory属性指示要使用哪个连接工厂来管理适配器的连接。
这channel属性指定消息到达出站适配器的通道以及入站适配器将消息放置在的通道上。
虽然入站适配器和出站适配器都可以共享连接工厂，但服务器连接工厂始终由入站适配器“拥有”。
客户端连接工厂始终由出站适配器“拥有”。
每种类型的适配器只能获得对连接工厂的引用。
以下示例显示了如何定义客户端和服务器 TCP 连接工厂：
<bean id="javaSerializer"
class="org.springframework.core.serializer.DefaultSerializer"/>
<bean id="javaDeserializer"
class="org.springframework.core.serializer.DefaultDeserializer"/>
<int-ip:tcp-connection-factory id="server"
type="server"
port="1234"
deserializer="javaDeserializer"
serializer="javaSerializer"
using-nio="true"
single-use="true"/>
<int-ip:tcp-connection-factory id="client"
type="client"
host="localhost"
port="#{server.port}"
single-use="true"
so-timeout="10000"
deserializer="javaDeserializer"
serializer="javaSerializer"/>
<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"/>
<int-ip:tcp-inbound-channel-adapter id="inboundServer"
channel="loop"
connection-factory="server"/>
<int-ip:tcp-outbound-channel-adapter id="outboundServer"
channel="loop"
connection-factory="server"/>
<int:channel id="loop"/>在前面的配置中，到达input通道通过client连接工厂，在服务器上接收，并放置在loop渠道。
因为loop是outboundServer，消息通过同一连接循环回传，由inboundClient，并存入replies渠道。
Java 序列化用于网络。
通常，入站适配器使用type="server"连接工厂，它监听传入的连接请求。
在某些情况下，您可能希望反向建立连接，以便入站适配器连接到外部服务器，然后等待该连接上的入站消息。
此拓扑由设置client-mode="true"在入站适配器上。
在这种情况下，连接工厂的类型必须是client并且必须有single-use设置为false.
另外两个属性支持此机制。
这retry-interval指定（以毫秒为单位）框架在连接失败后尝试重新连接的频率。
这scheduler供应TaskScheduler以安排连接尝试并测试连接是否仍然处于活动状态。
如果不提供调度程序，则使用框架的默认 taskScheduler bean。
对于出站适配器，通常在发送第一条消息时建立连接。
这client-mode="true"在出站适配器上导致在启动适配器时建立连接。
默认情况下，适配器会自动启动。
同样，连接工厂的类型必须是client并且有single-use="false".
一个retry-interval和scheduler也受支持。
如果连接失败，则由调度程序或在发送下一条消息时重新建立连接。
对于入站和出站，如果适配器已启动，则可以通过发送<control-bus />命令：@adapter_id.retryConnection().
然后你可以检查当前状态@adapter_id.isClientModeConnected().
TCP 网关
入站 TCP 网关TcpInboundGateway和出站 TCP 网关TcpOutboundGateway分别使用服务器和客户端连接工厂。
每个连接一次可以处理一个请求或响应。
入站网关在构造包含传入有效负载的消息并将其发送到requestChannel，等待响应并通过将响应消息写入连接来发送响应消息中的有效负载。
对于入站网关，您必须保留或填充ip_connectionId标头，因为它用于将消息与连接相关联。
源自网关的邮件会自动设置标头。
如果回复构造为新消息，则需要设置标头。
可以从传入消息中捕获标头值。 |
与入站适配器一样，入站网关通常使用type="server"连接工厂，它监听传入的连接请求。
在某些情况下，您可能希望反向建立连接，以便入站网关连接到外部服务器，然后等待并回复该连接上的入站消息。
此拓扑通过使用client-mode="true"在入站网关上。
在这种情况下，连接工厂的类型必须是client并且必须有single-use设置为false.
另外两个属性支持此机制。
这retry-interval指定（以毫秒为单位）框架在连接失败后尝试重新连接的频率。
这scheduler供应TaskScheduler以安排连接尝试并测试连接是否仍然处于活动状态。
如果网关已启动，您可以通过发送<control-bus/>命令：@adapter_id.retryConnection()并检查当前状态@adapter_id.isClientModeConnected().
出站网关通过连接发送消息后，等待响应，构造响应消息，并将其放在回复通道上。 通过连接进行的通信是单线程的。 一次只能处理一条消息。 如果另一个线程在收到当前响应之前尝试发送消息，它会阻塞，直到任何以前的请求完成（或超时）。 但是，如果将客户机连接工厂配置为一次性连接，则每个新请求都会获得自己的连接并立即处理。 以下示例配置入站 TCP 网关：
<int-ip:tcp-inbound-gateway id="inGateway"
request-channel="tcpChannel"
reply-channel="replyChannel"
connection-factory="cfServer"
reply-timeout="10000"/>如果使用配置了缺省序列化程序或反序列化程序的连接工厂，则消息为\r\n分隔数据和网关可以由简单的客户端（如 telnet）使用。
以下示例显示了出站 TCP 网关：
<int-ip:tcp-outbound-gateway id="outGateway"
request-channel="tcpChannel"
reply-channel="replyChannel"
connection-factory="cfClient"
request-timeout="10000"
remote-timeout="10000"/> <!-- or e.g. remote-timeout-expression="headers['timeout']" -->这client-mode当前不适用于出站网关。
从版本 5.2 开始，可以使用属性closeStreamAfterSend.
如果连接工厂配置为single-use（每个请求/回复的新连接）网关将关闭输出流;这会向服务器发出 EOF 信号。
如果服务器使用 EOF 来确定消息的结尾，而不是流中的某个分隔符，但保持连接打开以接收回复，这将很有用。
通常，调用线程会阻塞在网关中，等待回复（或超时）。
从 5.3 版开始，您可以将async网关上的属性，并且释放发送线程以执行其他工作。
回复（或错误）将在接收线程上发送。
这仅适用于使用TcpNetClientConnectionFactory，在使用 NIO 时将其忽略，因为存在竞争条件，即在回复之前，接收回复后发生的套接字错误可能会传递给网关。
使用共享连接时 （singleUse=false），当另一个请求正在处理时，将阻止一个新请求，直到收到当前回复。
考虑使用CachingClientConnectionFactory如果您希望支持长期连接池上的并发请求。 |
从版本 5.4 开始，可以配置入站unsolicitedMessageChannel.
未经请求的入站消息将发送到此通道，以及延迟回复（客户端超时）。
为了在服务器端支持这一点，您现在可以注册多个TcpSenders 与连接工厂。
网关和通道适配器会自动注册。
从服务器发送未经请求的消息时，必须添加适当的IpHeaders.CONNECTION_ID发送的消息。
TCP 消息关联
IP 端点的一个目标是提供与 Spring Integration 应用程序以外的系统的通信。 因此，默认情况下仅发送和接收消息有效负载。 从 3.0 开始，您可以使用 JSON、Java 序列化或自定义序列化程序和反序列化程序来传输标头。 有关详细信息，请参阅传输标头。 框架（除非在服务器端使用网关）或协作通道适配器提供消息关联。在本文档的后面，我们将讨论可用于应用程序的各种关联技术。 在大多数情况下，这需要消息的特定应用程序级关联，即使消息有效负载包含一些自然关联数据（例如订单号）。
网关
网关会自动关联消息。 但是，对于相对较低容量的应用程序，应使用出站网关。 将连接工厂配置为对所有消息对使用单个共享连接 （'single-use=“false”'） 时，一次只能处理一条消息。 新消息必须等到收到对上一条消息的回复。 当为每条新消息配置连接工厂以使用新连接 （'single-use=“true”'） 时，此限制不适用。 虽然此设置可以提供比共享连接环境更高的吞吐量，但它会带来为每个消息对打开和关闭新连接的开销。
因此，对于大容量消息，请考虑使用一对协作通道适配器。 但是，为此，您需要提供协作逻辑。
Spring Integration 2.2 中引入的另一个解决方案是使用CachingClientConnectionFactory，这允许使用共享连接池。
协作出站和入站通道适配器
若要实现高容量吞吐量（避免使用网关的陷阱，如前所述），可以配置一对协作的出站和入站通道适配器。 您还可以使用协作适配器（服务器端或客户端）进行完全异步通信（而不是使用请求-回复语义）。 在服务器端，消息关联由适配器自动处理，因为入站适配器添加了一个标头，允许出站适配器确定在发送回复消息时要使用的连接。
在服务器端，您必须填充ip_connectionId标头，因为它用于将消息与连接相关联。
源自入站适配器的邮件会自动设置标头。
如果要构造其他消息以发送，则需要设置标头。
可以从传入邮件中获取标头值。 |
在客户端，应用程序必须根据需要提供自己的关联逻辑。 您可以通过多种方式做到这一点。
如果消息有效负载具有一些自然关联数据（例如事务 ID 或订单号），并且您不需要保留原始出站消息中的任何信息（例如回复通道标头），则关联很简单，并且在任何情况下都会在应用程序级别完成。
如果消息有效负载具有一些自然关联数据（例如事务标识或订单号），但您需要保留原始出站消息中的某些信息（例如应答通道头），则可以保留原始出站消息的副本（可能使用发布-订阅通道）并使用聚合器重新组合必要的数据。
对于前两种方案中的任何一种，如果有效负载没有自然关联数据，则可以在出站通道适配器的上游提供转换器，以使用此类数据增强有效负载。 这样的转换器可以将原始有效负载转换为包含原始有效负载和消息标头的某些子集的新对象。 当然，标头中的实时对象（例如回复通道）不能包含在转换后的有效负载中。
如果选择此类策略，则需要确保连接工厂具有适当的序列化程序-解序列化程序对来处理此类有效负载（例如DefaultSerializer和DefaultDeserializer，它使用 Java 序列化，或自定义序列化器和反序列化程序）。
这ByteArray*SerializerTCP 连接工厂中提到的选项，包括默认的ByteArrayCrLfSerializer，则不支持此类有效负载，除非转换后的有效负载是String或byte[].
|
在 2.2 版本之前，当协作通道适配器使用客户端连接工厂时， 此默认行为不适用于真正的异步环境，因此它现在默认为无限超时。
您可以通过将 |
从 V5.4 开始，多个出站通道适配器和一个TcpInboundChannelAdapter可以共享同一个连接工厂。
这允许应用程序同时支持请求/回复和任意服务器→客户端消息传递。
有关详细信息，请参阅 TCP 网关。
传输标头
TCP 是一种流协议。Serializers和Deserializers在流中划分消息。
在 3.0 之前，只有消息有效负载 （String或byte[]）可以通过 TCP 传输。
从 3.0 开始，您可以传输选定的标头以及有效负载。
但是，“活动”对象，例如replyChannelheader，无法序列化。
通过 TCP 发送标头信息需要一些额外的配置。
第一步是提供ConnectionFactory使用MessageConvertingTcpMessageMapper使用mapper属性。
此映射器委托给任何MessageConverter实现将消息转换为某个对象或从某个对象转换消息，这些对象可以由配置的serializer和deserializer.
Spring Integration 提供了一个MapMessageConverter，它允许指定添加到Map对象以及有效负载。生成的 Map 有两个条目：payload和headers.
这headersentry 本身是一个Map并包含选定的标头。
第二步是提供一个序列化程序和一个反序列化程序，可以在Map和一些线格式。这可以是自定义的Serializer或Deserializer，如果对等系统不是 Spring Integration 应用程序，您通常需要它。
Spring Integration 提供了一个MapJsonSerializer将Map与 JSON 之间。它使用 Spring IntegrationJsonObjectMapper. 可以提供自定义JsonObjectMapper如果需要。默认情况下，序列化程序会插入换行符 （0x0a） 字符。有关更多信息，请参阅 Javadoc。
这JsonObjectMapper使用任何版本的Jackson在类路径上。 |
您还可以使用Map，通过使用DefaultSerializer和DefaultDeserializer.
以下示例显示了传输correlationId,sequenceNumber和sequenceSize标头：
<int-ip:tcp-connection-factory id="client"
type="client"
host="localhost"
port="12345"
mapper="mapper"
serializer="jsonSerializer"
deserializer="jsonSerializer"/>
<bean id="mapper"
class="o.sf.integration.ip.tcp.connection.MessageConvertingTcpMessageMapper">
<constructor-arg name="messageConverter">
<bean class="o.sf.integration.support.converter.MapMessageConverter">
<property name="headerNames">
<list>
<value>correlationId</value>
<value>sequenceNumber</value>
<value>sequenceSize</value>
</list>
</property>
</bean>
</constructor-arg>
</bean>
<bean id="jsonSerializer" class="o.sf.integration.ip.tcp.serializer.MapJsonSerializer" />使用上述配置发送的消息，有效负载为“something”，将显示在网络上，如下所示：
{"headers":{"correlationId":"things","sequenceSize":5,"sequenceNumber":1},"payload":"something"}关于无阻塞 I/O （NIO）
使用蔚来（参见using-nio在 IP 配置属性中）避免专用于从每个套接字读取线程。
对于少数套接字，您可能会发现不使用 NIO，以及异步切换（例如QueueChannel），性能与使用蔚来一样好或更好。
在处理大量连接时，您应该考虑使用蔚来汽车。 然而，蔚来汽车的使用还有一些其他影响。 线程池（在任务执行器中）在所有套接字之间共享。 每个传入消息都被组装并作为从该池中选择的线程上的单独工作单元发送到配置的通道。 到达同一套接字的两条顺序消息可能由不同的线程处理。 这意味着消息发送到通道的顺序是不确定的。 不维护到达套接字的消息的严格排序。
对于某些应用程序，这不是问题。
对于其他人来说，这是一个问题。
如果您需要严格排序，请考虑将using-nio自false以及使用异步切换。
或者，您可以在入站端点的下游插入重排序器，以将消息返回到正确的顺序。
如果您将apply-sequence自true在连接工厂上，到达 TCP 连接的消息具有sequenceNumber和correlationIdheaders 设置。
重排序器使用这些标头将消息返回到正确的顺序。
从版本 5.1.4 开始，接受新连接优先于从现有连接读取。
通常，除非新传入连接的速率非常高，否则这应该影响不大。
如果您希望恢复到之前赋予读取优先级的行为，请将multiAccept属性TcpNioServerConnectionFactory自false. |
池大小
不再使用池大小属性。
以前，当未指定任务执行器时，它会指定默认线程池的大小。
它还用于在服务器套接字上设置连接积压。
不再需要第一个功能（见下一段）。
第二个函数被backlog属性。
以前，当将固定线程池任务执行器（这是默认值）与 NIO 一起使用时，可能会出现死锁并且处理将停止。 当缓冲区已满、从套接字读取的线程尝试向缓冲区添加更多数据，并且没有线程可用于在缓冲区中腾出空间时，会出现此问题。 这只发生在池子面积非常小的情况下，但在极端条件下可能是可能的。 从 2.2 开始，有两个变化消除了这个问题。 首先，默认任务执行器是缓存的线程池执行器。 其次，添加了死锁检测逻辑，如果发生线程匮乏，则抛出异常而不是死锁，从而释放死锁资源。
| 现在，默认任务执行器是无界的，如果消息处理需要较长时间，则可能会在传入消息速率较高的情况下出现内存不足情况。 如果应用程序表现出此类行为，则应使用具有适当池大小的池化任务执行器，但请参阅下一节。 |
线程池任务执行器CALLER_RUNS政策
当您将固定线程池与CallerRunsPolicy (CALLER_RUNS使用<task/>命名空间），队列容量较小。
如果不使用固定线程池，则以下内容不适用。
对于蔚来连接，有三种不同的任务类型。 I/O 选择器处理在一个专用线程上执行（检测事件、接受新连接以及使用任务执行器将 I/O 读取作分派到其他线程）。 当 I/O 读取器线程 （读取作被分派到） 读取数据时，它会移交给另一个线程来组合传入消息。 大型消息可能需要多次读取才能完成。 这些“汇编器”线程可能会在等待数据时阻塞。 当发生新的读取事件时，读取器会确定此套接字是否已经具有汇编程序，如果没有，则运行一个新的汇编程序。 汇编过程完成后，汇编程序线程将返回到池中。
当池耗尽时，这可能会导致死锁，CALLER_RUNS拒绝策略正在使用中，并且任务队列已满。
当池为空且队列中没有空间时，IO 选择器线程会收到OP_READ事件并使用 executor 调度读取。
队列已满，因此选择器线程本身启动读取过程。
现在，它检测到此套接字没有汇编程序，并在读取之前触发汇编程序。
同样，队列已满，选择器线程成为汇编器。
汇编程序现在被阻止，等待读取数据，这永远不会发生。
连接工厂现在死锁，因为选择器线程无法处理新事件。
为了避免这种死锁，我们必须避免执行汇编任务的选择器（或读取器）线程。 我们希望将单独的池用于 IO 和汇编作。
该框架提供了一个CompositeExecutor，它允许配置两个不同的执行器：一个用于执行 IO作，一个用于消息汇编。
在此环境中，IO 线程永远无法成为汇编器线程，并且不会发生死锁。
此外，任务执行器应配置为使用AbortPolicy (ABORT使用时<task>).
当 I/O 任务无法完成时，它会延迟一小段时间，并不断重试，直到可以完成并分配汇编程序。
默认情况下，延迟为 100 毫秒，但您可以通过设置readDelay连接工厂 （read-delay使用 XML 命名空间进行配置时）。
以下三个示例展示了如何配置复合执行器：
@Bean
private CompositeExecutor compositeExecutor() {
ThreadPoolTaskExecutor ioExec = new ThreadPoolTaskExecutor();
ioExec.setCorePoolSize(4);
ioExec.setMaxPoolSize(10);
ioExec.setQueueCapacity(0);
ioExec.setThreadNamePrefix("io-");
ioExec.setRejectedExecutionHandler(new AbortPolicy());
ioExec.initialize();
ThreadPoolTaskExecutor assemblerExec = new ThreadPoolTaskExecutor();
assemblerExec.setCorePoolSize(4);
assemblerExec.setMaxPoolSize(10);
assemblerExec.setQueueCapacity(0);
assemblerExec.setThreadNamePrefix("assembler-");
assemblerExec.setRejectedExecutionHandler(new AbortPolicy());
assemblerExec.initialize();
return new CompositeExecutor(ioExec, assemblerExec);
}
<bean id="myTaskExecutor" class="org.springframework.integration.util.CompositeExecutor">
<constructor-arg ref="io"/>
<constructor-arg ref="assembler"/>
</bean>
<task:executor id="io" pool-size="4-10" queue-capacity="0" rejection-policy="ABORT" />
<task:executor id="assembler" pool-size="4-10" queue-capacity="0" rejection-policy="ABORT" /><bean id="myTaskExecutor" class="org.springframework.integration.util.CompositeExecutor">
<constructor-arg>
<bean class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
<property name="threadNamePrefix" value="io-" />
<property name="corePoolSize" value="4" />
<property name="maxPoolSize" value="8" />
<property name="queueCapacity" value="0" />
<property name="rejectedExecutionHandler">
<bean class="java.util.concurrent.ThreadPoolExecutor.AbortPolicy" />
</property>
</bean>
</constructor-arg>
<constructor-arg>
<bean class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
<property name="threadNamePrefix" value="assembler-" />
<property name="corePoolSize" value="4" />
<property name="maxPoolSize" value="10" />
<property name="queueCapacity" value="0" />
<property name="rejectedExecutionHandler">
<bean class="java.util.concurrent.ThreadPoolExecutor.AbortPolicy" />
</property>
</bean>
</constructor-arg>
</bean>SSL/TLS 支持
支持安全套接字层/传输层安全性。
使用蔚来时，JDK 5+SSLEngine功能用于在建立连接后处理握手。
不使用蔚来时，标准SSLSocketFactory和SSLServerSocketFactory对象用于创建连接。
提供了许多策略接口以允许进行大量定制。
这些接口的默认实现提供了开始使用安全通信的最简单方法。
开始
无论是否使用蔚来，都需要配置ssl-context-support属性。
此属性引用描述所需密钥存储的位置和密码的 <bean/> 定义。
SSL/TLS 对等体每个需要两个密钥存储：
-
包含用于标识对等方的私钥和公钥对的密钥库
-
一个信任库，其中包含受信任对等方的公钥。 请参阅文档
keytoolJDK 提供的实用程序。 基本步骤是-
创建一个新的密钥对并将其存储在密钥库中。
-
导出公钥。
-
将公密钥导入对等方的信任库。
-
对另一个对等方重复上述步骤。
-
| 在测试用例中，在两个对等节点上使用相同的密钥存储是很常见的，但在生产中应避免这种情况。 |
建立密钥存储后，下一步是向TcpSSLContextSupportbean 并向连接工厂提供对该 bean 的引用。
以下示例配置 SSL 连接：
<bean id="sslContextSupport"
class="o.sf.integration.ip.tcp.connection.support.DefaultTcpSSLContextSupport">
<constructor-arg value="client.ks"/>
<constructor-arg value="client.truststore.ks"/>
<constructor-arg value="secret"/>
<constructor-arg value="secret"/>
</bean>
<ip:tcp-connection-factory id="clientFactory"
type="client"
host="localhost"
port="1234"
ssl-context-support="sslContextSupport" />这DefaultTcpSSLContextSupportclass 也有一个可选的protocol属性，可以是SSL或TLS（默认值）。
密钥库文件名（前两个构造函数参数）使用 SpringResource抽象化。
默认情况下，这些文件位于类路径上，但您可以使用file:prefix （而是在文件系统上查找文件）。
从 4.3.6 版本开始，当您使用 NIO 时，您可以指定ssl-handshake-timeout（以秒为单位）在连接工厂上。
此超时（默认值为 30 秒）在等待数据时的 SSL 握手期间使用。
如果超过超时，则停止进程并关闭套接字。
主机验证
从 5.0.8 版本开始，您可以配置是否启用主机验证。 从 5.1 版本开始，默认启用;禁用它的机制取决于您是否使用蔚来。
主机验证用于确保您连接到的服务器与证书中的信息匹配，即使证书是可信的。
使用蔚来时，配置DefaultTcpNioSSLConnectionSupport例如。
@Bean
public DefaultTcpNioSSLConnectionSupport connectionSupport() {
DefaultTcpSSLContextSupport sslContextSupport = new DefaultTcpSSLContextSupport("test.ks",
"test.truststore.ks", "secret", "secret");
sslContextSupport.setProtocol("SSL");
DefaultTcpNioSSLConnectionSupport tcpNioConnectionSupport =
new DefaultTcpNioSSLConnectionSupport(sslContextSupport, false);
return tcpNioConnectionSupport;
}
第二个构造函数参数禁用主机验证。
这connectionSupport然后将 bean 注入 NIO 连接工厂。
不使用蔚来时，配置在TcpSocketSupport:
connectionFactory.setTcpSocketSupport(new DefaultTcpSocketSupport(false));
同样，构造函数参数禁用主机验证。
先进技术
本节介绍在某些情况下可能会发现有用的高级技术。
策略接口
在许多情况下，前面描述的配置是启用通过 TCP/IP 进行安全通信所需的全部内容。 但是，Spring Integration 提供了许多策略接口来允许自定义和修改套接字工厂和套接字：
-
TcpSSLContextSupport -
TcpSocketFactorySupport -
TcpSocketSupport -
TcpNetConnectionSupport -
TcpNioConnectionSupport
这TcpSSLContextSupport策略界面
以下列表显示了TcpSSLContextSupport策略接口：
public interface TcpSSLContextSupport {
SSLContext getSSLContext() throws Exception;
}
的实现TcpSSLContextSupport接口负责创建一个SSLContext对象。
框架提供的实现是DefaultTcpSSLContextSupport，前面描述。
如果您需要不同的行为，请实现此接口，并为连接工厂提供对类实现的 bean 的引用。
这TcpSocketFactorySupport策略界面
以下列表显示了TcpSocketFactorySupport策略接口：
public interface TcpSocketFactorySupport {
ServerSocketFactory getServerSocketFactory();
SocketFactory getSocketFactory();
}
此接口的实现负责获取对ServerSocketFactory和SocketFactory.
提供了两种实现。
首先是DefaultTcpNetSocketFactorySupport对于非 SSL 套接字（当没有ssl-context-support属性）。
这使用 JDK 的默认工厂。
第二个实现是DefaultTcpNetSSLSocketFactorySupport.
默认情况下，当ssl-context-support属性。
它使用SSLContext由该 bean 创建以创建套接字工厂。
此接口仅适用于以下情况using-nio是false.
蔚来汽车不使用插座工厂。 |
这TcpSocketSupport策略界面
以下列表显示了TcpSocketSupport策略接口：
public interface TcpSocketSupport {
void postProcessServerSocket(ServerSocket serverSocket);
void postProcessSocket(Socket socket);
}
此接口的实现可以在创建套接字后以及应用所有配置的属性之后但在使用套接字之前修改套接字。
无论您是否使用蔚来汽车，这都适用。
例如，您可以使用此接口的实现来修改 SSL 套接字上支持的密码套件，或者可以添加一个侦听器，该侦听器在 SSL 握手完成后收到通知。
该框架提供的唯一实现是DefaultTcpSocketSupport，它不会以任何方式修改套接字。
提供您自己的实现TcpSocketFactorySupport或TcpSocketSupport，通过设置socket-factory-support和socket-support属性。
这TcpNetConnectionSupport策略界面
以下列表显示了TcpNetConnectionSupport策略接口：
public interface TcpNetConnectionSupport {
TcpNetConnection createNewConnection(Socket socket,
boolean server, boolean lookupHost,
ApplicationEventPublisher applicationEventPublisher,
String connectionFactoryName) throws Exception;
}
调用此接口以创建类型TcpNetConnection（或其子类）。
该框架提供了一个单一的实现（DefaultTcpNetConnectionSupport），默认情况下，它会创建简单的TcpNetConnection对象。
它有两个属性：pushbackCapable和pushbackBufferSize.
启用后推后，实现返回一个子类，该子类包装连接的InputStream在PushbackInputStream.
与PushbackInputStream默认，缓冲区大小默认为 1。
这允许反序列化器将字节“未读”（推回）到流中。
以下简单示例演示了如何在委托反序列化程序中使用它，该委托反序列化程序“查看”第一个字节以确定要调用的反序列化程序：
public class CompositeDeserializer implements Deserializer<byte[]> {
private final ByteArrayStxEtxSerializer stxEtx = new ByteArrayStxEtxSerializer();
private final ByteArrayCrLfSerializer crlf = new ByteArrayCrLfSerializer();
@Override
public byte[] deserialize(InputStream inputStream) throws IOException {
PushbackInputStream pbis = (PushbackInputStream) inputStream;
int first = pbis.read();
if (first < 0) {
throw new SoftEndOfStreamException();
}
pbis.unread(first);
if (first == ByteArrayStxEtxSerializer.STX) {
this.receivedStxEtx = true;
return this.stxEtx.deserialize(pbis);
}
else {
this.receivedCrLf = true;
return this.crlf.deserialize(pbis);
}
}
}
这TcpNioConnectionSupport策略界面
以下列表显示了TcpNioConnectionSupport策略接口：
public interface TcpNioConnectionSupport {
TcpNioConnection createNewConnection(SocketChannel socketChannel,
boolean server, boolean lookupHost,
ApplicationEventPublisher applicationEventPublisher,
String connectionFactoryName) throws Exception;
}
调用此接口以创建TcpNioConnection对象（或子类中的对象）。
Spring Integration 提供了两种实现：DefaultTcpNioSSLConnectionSupport和DefaultTcpNioConnectionSupport.
使用哪一个取决于是否正在使用 SSL。
一个常见的用例是将DefaultTcpNioSSLConnectionSupport并覆盖postProcessSSLEngine.
请参阅 SSL 客户端身份验证示例。
与DefaultTcpNetConnectionSupport，这些实现也支持 push back。
示例：启用 SSL 客户端身份验证
要在使用 SSL 时启用客户端证书身份验证，该技术取决于您是否使用 NIO。
当你不NIO时，提供自定义TcpSocketSupport实现对服务器套接字进行后处理：
serverFactory.setTcpSocketSupport(new DefaultTcpSocketSupport() {
@Override
public void postProcessServerSocket(ServerSocket serverSocket) {
((SSLServerSocket) serverSocket).setNeedClientAuth(true);
}
});
（当您使用 XML 配置时，通过设置socket-support属性）。
当您使用蔚来时，请提供自定义TcpNioSslConnectionSupport实现以后处理SSLEngine，如以下示例所示：
@Bean
public DefaultTcpNioSSLConnectionSupport tcpNioConnectionSupport() {
return new DefaultTcpNioSSLConnectionSupport(serverSslContextSupport) {
@Override
protected void postProcessSSLEngine(SSLEngine sslEngine) {
sslEngine.setNeedClientAuth(true);
}
}
}
@Bean
public TcpNioServerConnectionFactory server() {
...
serverFactory.setTcpNioConnectionSupport(tcpNioConnectionSupport());
...
}
（当您使用 XML 配置时，从 4.3.7 版开始，通过将nio-connection-support属性）。
IP 配置属性
下表描述了可以设置以配置 IP 连接的属性：
| 属性名称 | 客户？ | 服务器？ | 允许的值 | 属性描述 |
|---|---|---|---|---|
|
Y |
Y |
客户端、服务器 |
确定连接工厂是客户机还是服务器。 |
|
Y |
N |
目标的主机名或 IP 地址。 |
|
|
Y |
Y |
端口。 |
|
|
Y |
Y |
实现 |
|
|
Y |
Y |
实现 |
|
|
Y |
Y |
|
连接是否使用蔚来。
请参阅 |
|
Y |
N |
|
使用蔚来时，连接是否使用直接缓冲区。
请参阅 |
|
Y |
Y |
|
当您使用蔚来汽车时，可能需要对消息进行重新排序。
当此属性设置为 |
|
Y |
Y |
默认为 |
|
|
Y |
Y |
看 |
|
|
Y |
Y |
看 |
|
|
Y |
Y |
|
看 |
|
Y |
Y |
集 |
|
|
Y |
Y |
|
看 |
|
Y |
Y |
看 |
|
|
N |
Y |
在多宿主系统上，为套接字绑定到的接口指定 IP 地址。 |
|
|
Y |
Y |
指定要用于套接字处理的特定执行器。
如果未提供，则使用内部缓存线程执行器。
在某些需要使用特定任务执行器的平台上需要，例如 |
|
|
Y |
Y |
|
指定一个连接是否可以用于多条消息。
如果 |
|
N |
N |
不再使用此属性。
为了向后兼容，它会设置积压工作，但您应该使用 |
|
|
N |
Y |
设置服务器工厂的连接积压工作。 |
|
|
Y |
Y |
|
指定是否对 IP 地址进行反向查找，以转换为主机名以用于邮件头。
如果为 false，则改用 IP 地址。
违约： |
|
Y |
Y |
请参阅 TCP 连接拦截器。 |
|
|
Y |
Y |
||
|
Y |
Y |
||
|
Y |
Y |
请参阅 SSL/TLS 支持。 |
|
|
Y |
Y |
请参阅高级技术。 |
|
|
Y |
Y |
长> 0 |
由于线程不足，上一次尝试失败后重试读取之前的延迟（以毫秒为单位）。
默认值：100。
仅适用于以下情况 |
下表描述了可以设置的属性以配置 UDP 入站通道适配器：
| 属性名称 | 允许的值 | 属性描述 |
|---|---|---|
|
适配器侦听的端口。 |
|
|
|
UDP 适配器是否使用多播。 |
|
当多播为 true 时，适配器加入的多播地址。 |
|
|
指定可以并发处理的数据包数。 仅当未配置任务执行器时才适用。 默认值：5。 |
|
任务执行器 |
指定要用于套接字处理的特定执行器。
如果未提供，则使用内部池执行器。
在某些需要使用特定任务执行器的平台上需要，例如 |
|
|
用于接收的缓冲区的大小 |
|
|
|
UDP 适配器是否期望接收到的数据包中有一个数据长度字段。 用于检测数据包截断。 |
|
请参阅 |
|
|
用于 UDP 确认数据包。
请参阅中的 setSendBufferSize（） 方法 |
|
|
看 |
|
|
在多宿主系统上，为套接字绑定到的接口指定 IP 地址。 |
|
|
如果下游组件抛出异常，则 |
|
|
|
指定是否对 IP 地址进行反向查找，以转换为主机名以用于邮件头。
如果 |
下表描述了可以设置的属性以配置 UDP 出站通道适配器：
| 属性名称 | 允许的值 | 属性描述 |
|---|---|---|
|
目标的主机名或 IP 地址。 对于多播 udp 适配器，多播地址。 |
|
|
目标上的端口。 |
|
|
|
udp 适配器是否使用多播。 |
|
|
UDP 适配器是否需要来自目标的确认。
启用后，需要设置以下四个属性： |
|
什么时候 |
|
|
什么时候 |
|
|
什么时候 |
|
|
默认为 1。 对于多播适配器，可以将此值设置为较大的值，这需要来自多个目标的确认。 |
|
|
|
UDP 适配器是否在发送到目标的数据包中包含数据长度字段。 |
|
对于多播适配器，指定 |
|
|
看 |
|
|
请参阅 |
|
|
用于 UDP 确认数据包。
请参阅 |
|
本地地址 |
在多宿主系统上，对于 UDP 适配器，为套接字绑定到的应答消息的接口指定 IP 地址。 对于组播适配器，它还确定通过哪个接口发送组播数据包。 |
|
|
指定要用于确认处理的特定执行程序。
如果未提供，则使用内部单线程执行器。
在某些需要使用特定任务执行器的平台上需要，例如 |
|
|
SpEL表达 |
要评估的 SpEL 表达式以确定哪个 |
|
SpEL表达 |
要计算的 SpEL 表达式，以确定哪个数据报套接字用于发送传出 UDP 数据包。 |
下表描述了可以设置的属性以配置 TCP 入站通道适配器：
| 属性名称 | 允许的值 | 属性描述 |
|---|---|---|
|
将入站消息发送到的通道。 |
|
|
如果连接工厂的类型为 |
|
|
如果下游组件抛出异常，则 |
|
|
|
什么时候 |
|
当在 |
|
|
|
指定一个 |
下表描述了可以设置的属性以配置 TCP 出站通道适配器：
| 属性名称 | 允许的值 | 属性描述 |
|---|---|---|
|
出站消息到达的通道。 |
|
|
如果连接工厂的类型为 |
|
|
|
什么时候 |
|
当在 |
|
|
|
指定一个 |
下表描述了您可以设置的属性来配置 TCP 入站网关：
| 属性名称 | 允许的值 | 属性描述 |
|---|---|---|
|
连接工厂必须是服务器类型。 |
|
|
传入消息发送到的通道。 |
|
|
回复消息可能到达的通道。 通常，回复到达添加到入站邮件标头的临时回复通道。 |
|
|
网关等待回复的时间（以毫秒为单位）。 默认值：1000（1 秒）。 |
|
|
如果下游组件抛出异常，则 |
|
|
|
什么时候 |
|
当在 |
|
|
|
指定一个 |
下表描述了可以设置以配置 TCP 出站网关的属性：
| 属性名称 | 允许的值 | 属性描述 |
|---|---|---|
|
连接工厂必须是 |
|
|
传出邮件到达的通道。 |
|
|
自选。 将回复消息发送到的通道。 |
|
|
网关等待来自远程系统的应答的时间（以毫秒为单位）。
相互排斥 |
|
|
根据消息评估的 SpEL 表达式，以确定网关等待来自远程系统的应答的时间（以毫秒为单位）。
相互排斥 |
|
|
如果未使用一次性连接工厂，则网关等待访问共享连接的时间（以毫秒为单位）。 |
|
|
网关在将回复发送到回复通道时等待的时间（以毫秒为单位）。 仅当应答通道可能阻塞（例如当前已满的有界 QueueChannel）时才适用。 |
|
|
发送后释放发送线程;回复（或错误）将在接收线程上发送。 |
|
|
向其发送未经请求的消息和延迟回复的通道。 |
IP 消息头
本模块使用以下内容MessageHeader实例：
| 标头名称 | IpHeaders 常量 | 描述 |
|---|---|---|
|
|
从中接收 TCP 消息或 UDP 数据包的主机名。
如果 |
|
|
从中接收 TCP 消息或 UDP 数据包的 IP 地址。 |
|
|
UDP 数据包的远程端口。 |
ip_localInetAddress |
|
本地 |
|
|
将 UDP 应用程序级确认发送到的远程 IP 地址。 该框架在数据包中包含确认信息。 |
|
|
UDP 应用程序级确认的相关 ID。 该框架在数据包中包含确认信息。 |
|
|
TCP 连接的远程端口。 |
|
|
TCP 连接的唯一标识符。 由入站消息的框架设置。 发送到服务器端入站通道适配器或回复入站网关时，需要此标头，以便终结点可以确定要将消息发送到的连接。 |
|
|
仅供参考。 使用缓存或故障转移客户端连接工厂时，它包含实际的基础连接 ID。 |
|
|
入站邮件的可选内容类型
在此表后描述。
请注意，与其他标头常量不同，此常量位于 |
对于入站消息，ip_hostname,ip_address,ip_tcp_remotePort和ip_connectionId默认映射TcpHeaderMapper.
如果将映射器的addContentTypeHeader属性设置为true，映射器将contentType标头 （application/octet-stream;charset="UTF-8"，默认情况下）。
您可以通过设置contentType财产。
您可以通过子类化来添加其他标头TcpHeaderMapper并覆盖supplyCustomHeaders方法。
例如，当您使用 SSL 时，您可以添加SSLSession通过从TcpConnection对象，它作为参数提供给supplyCustomHeaders方法。
对于出站邮件，String有效负载转换为byte[]使用默认值 （UTF-8） 字符集。
将charset属性来更改默认值。
在自定义映射器属性或子类化时，将映射器声明为 bean，并使用mapper财产。
基于注释的配置
示例存储库中的以下示例显示了使用注释而不是 XML 时可用的一些配置选项：
@EnableIntegration (1)
@IntegrationComponentScan (2)
@Configuration
public static class Config {
@Value(${some.port})
private int port;
@MessagingGateway(defaultRequestChannel="toTcp") (3)
public interface Gateway {
String viaTcp(String in);
}
@Bean
@ServiceActivator(inputChannel="toTcp") (4)
public MessageHandler tcpOutGate(AbstractClientConnectionFactory connectionFactory) {
TcpOutboundGateway gate = new TcpOutboundGateway();
gate.setConnectionFactory(connectionFactory);
gate.setOutputChannelName("resultToString");
return gate;
}
@Bean (5)
public TcpInboundGateway tcpInGate(AbstractServerConnectionFactory connectionFactory) {
TcpInboundGateway inGate = new TcpInboundGateway();
inGate.setConnectionFactory(connectionFactory);
inGate.setRequestChannel(fromTcp());
return inGate;
}
@Bean
public MessageChannel fromTcp() {
return new DirectChannel();
}
@MessageEndpoint
public static class Echo { (6)
@Transformer(inputChannel="fromTcp", outputChannel="toEcho")
public String convert(byte[] bytes) {
return new String(bytes);
}
@ServiceActivator(inputChannel="toEcho")
public String upCase(String in) {
return in.toUpperCase();
}
@Transformer(inputChannel="resultToString")
public String convertResult(byte[] bytes) {
return new String(bytes);
}
}
@Bean
public AbstractClientConnectionFactory clientCF() { (7)
return new TcpNetClientConnectionFactory("localhost", this.port);
}
@Bean
public AbstractServerConnectionFactory serverCF() { (8)
return new TcpNetServerConnectionFactory(this.port);
}
}
| 1 | 标准 Spring Integration 注释，为集成应用程序启用基础架构。 |
| 2 | 搜索@MessagingGateway接口。 |
| 3 | 流客户端的入口点。调用应用程序可以使用@Autowired为此Gatewaybean 并调用其方法。 |
| 4 | 出站终结点由MessageHandler以及包装它的消费者。在这种情况下，@ServiceActivator根据通道类型配置端点。 |
| 5 | 入站端点（在 TCP/UDP 模块中）都是消息驱动的，因此只需要声明为 simple@Bean实例。 |
| 6 | 此类提供了许多用于此示例流的 POJO 方法（@Transformer和@ServiceActivator在服务器端和@Transformer在客户端）。 |
| 7 | 客户端连接工厂。 |
| 8 | 服务器端连接工厂。 |
将 Java DSL 用于 TCP 组件
对 TCP 组件的 DSL 支持包括适配器和网关的规范，以及Tcpclass 使用工厂方法创建连接工厂 bean，以及TcpCodecs类，用于创建序列化器和反序列化程序。
有关更多信息，请参阅他们的 javadocs。
下面是使用 DSL 配置流的一些示例。
@Bean
public IntegrationFlow server() {
return IntegrationFlow.from(Tcp.inboundAdapter(Tcp.netServer(1234)
.deserializer(TcpCodecs.lengthHeader1())
.backlog(30))
.errorChannel("tcpIn.errorChannel")
.id("tcpIn"))
.transform(Transformers.objectToString())
.channel("tcpInbound")
.get();
}
@Bean
public IntegrationFlow client() {
return f -> f.handle(Tcp.outboundAdapter(Tcp.nioClient("localhost", 1234)
.serializer(TcpCodecs.lengthHeader1())));
}
@Bean
public IntegrationFlow server() {
return IntegrationFlow.from(Tcp.inboundGateway(Tcp.netServer(1234)
.deserializer(TcpCodecs.lengthHeader1())
.serializer(TcpCodecs.lengthHeader1())
.backlog(30))
.errorChannel("tcpIn.errorChannel")
.id("tcpIn"))
.transform(Transformers.objectToString())
.channel("tcpInbound")
.get();
}
@Bean
public IntegrationFlow client() {
return f -> f.handle(Tcp.outboundGateway(Tcp.nioClient("localhost", 1234)
.deserializer(TcpCodecs.lengthHeader1())
.serializer(TcpCodecs.lengthHeader1())));
}