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 时,您还应在 host 属性中提供多播地址。 |
UDP 是一种高效但不稳定的协议。Spring Integration 添加了两个属性来提高可靠性:check-length
和 acknowledge
。当 check-length
设置为 true
时,适配器会在消息数据之前添加一个长度字段(网络字节序的四个字节)。这使得接收方可以验证接收到的数据包的长度。如果接收系统使用的缓冲区太短无法容纳数据包,数据包可能会被截断。长度头提供了一种检测此问题的方法。
从 4.3 版本开始,您可以将 port
设置为 0
,在这种情况下,操作系统会选择端口。在适配器启动且 isListening()
返回 true
后,可以通过调用 getPort()
来发现所选端口。
从 5.3.3 版本开始,您可以添加一个 SocketCustomizer
bean 来修改创建后的 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 入站通道适配器。 |
当 multicast 为 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 地址转换为用于消息头的 hostname,可以通过将 lookup-host
属性设置为 true
来覆盖默认行为。
从 5.3.3 版本开始,您可以添加一个 SocketCustomizer
bean 来修改创建后的 DatagramSocket
。它会用于接收 socket 以及为发送确认创建的任何 socket。
入站 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
(意味着操作系统选择端口)时,这非常有用。如果您需要在启动连接到 socket 的其他进程之前等待,也可以使用它代替轮询 isListening()
。
高级出站配置
<int-ip:udp-outbound-channel-adapter>
(UnicastSendingMessageHandler
) 具有 destination-expression
和 socket-expression
选项。
您可以使用 destination-expression
作为硬编码的 host-port 对的运行时替代方案,以根据 requestMessage
(作为评估上下文的根对象)确定出站数据报包的目标地址。表达式必须评估为 URI
、URI 风格的 String
(参见 RFC-2396)或 SocketAddress
。您还可以使用入站的 IpHeaders.PACKET_ADDRESS
头作为此表达式。在框架中,当我们在 UnicastReceivingChannelAdapter
中接收数据报并将其转换为消息时,DatagramPacketMessageMapper
会填充此头。头的精确值是传入数据报的 DatagramPacket.getSocketAddress()
的结果。
使用 socket-expression
,出站通道适配器可以使用(例如)入站通道适配器的 socket 通过接收数据的同一端口发送数据报。这在我们的应用程序作为 UDP 服务器运行且客户端在网络地址转换 (NAT) 后面的场景中非常有用。此表达式必须评估为 DatagramSocket
。requestMessage
用作评估上下文的根对象。您不能将 socket-expression
参数与 multicast
和 acknowledge
参数一起使用。以下示例展示了如何配置一个 UDP 入站通道适配器,其中包含一个将其转换为大写并使用 socket 的 transformer
<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();
}