前言
基于我的上一篇文章《Spring Integration超详细解读》,相信大家对Spring Integration已经有了基本的认识。
因此本文中,着重讲解Spring Integration Ip的实际应用。
导入依赖
如果是在POM中,则导入以下依赖(由于spring-integration-ip实际会引入spring-integration的依赖,因此无需再添加相关的依赖了)
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-ip</artifactId>
<version>5.5.18</version>
</dependency>
如果是在gradle中,则导入以下依赖
compile "org.springframework.integration:spring-integration-ip:5.5.18"
配置之前
Spring Integration Ip提供了对TCP和UDP的支持。
其中TcpSendingMessageHandler类用于发送TCP消息,TcpReceivingChannelAdapter类用于配置接收TCP消息的相关参数。
而UnicastSendingMessageHandler类用于发送UDP单播消息,UnicastReceivingChannelAdapter类用于配置接收UDP单播消息的相关参数。
另外,MulticastSendingMessageHandler类用于发送UDP组播消息,MulticastReceivingChannelAdapter类用于配置接收UDP组播消息的相关参数。
开始配置
UDP
以UDP为例,官方支持三种配置方式,XML配置方式、Java配置方式、Java DSL配置方式
UDP Outbound
UDP Outbound(UDP 出站)就是udp发送消息。以单播为例。
XML方式
<int-ip:udp-outbound-channel-adapter id="udpOut"
host="somehost"
port="11111"
multicast="false"
socket-customizer="udpCustomizer"
channel="exampleChannel"/>
Java方式
@Bean
public UnicastSendingMessageHandler handler() {
return new UnicastSendingMessageHandler("localhost", 11111);
}
Java DSL方式
@Bean
public IntegrationFlow udpOutFlow() {
return f -> f.handle(Udp.outboundAdapter("localhost", 1234)
.configureSocket(socket -> socket.setTrafficClass(0x10)))
.get();
}
UDP Inbound
UDP Inbound(UDP 入站)就是udp接收消息。以独播为例。
XML方式
<int-ip:udp-inbound-channel-adapter id="udpReceiver"
channel="udpOutChannel"
port="11111"
receive-buffer-size="500"
multicast="false"
socket-customizer="udpCustomizer"
check-length="true"/>
Java方式
@Bean
public UnicastReceivingChannelAdapter udpIn() {
UnicastReceivingChannelAdapter adapter = new UnicastReceivingChannelAdapter(11111);
adapter.setOutputChannelName("udpChannel");
return adapter;
}
Java DSL方式
@Bean
public IntegrationFlow udpIn() {
return IntegrationFlows.from(Udp.inboundAdapter(11111))
.channel("udpChannel")
.get();
}
项目实战
本文以UDP为例,分别配置客户端与服务端
客户端
定义用于配置UDP参数的类型
定义基础UDP参数类型
@Data
@NoArgsConstructor
@AllArgsConstructor
public class BaseUdpProp {
private String host;
private Integer port;
}
定义用于配置UDP单播入站参数的类型
@Component
@ConfigurationProperties(prefix = "custom.udp.unicast.inbound")
public class UnicastInboundUdpProp extends BaseUdpProp {
}
定义用于配置UDP单播出站参数的类型
@Component
@ConfigurationProperties(prefix = "custom.udp.unicast.outbound")
public class UnicastOutboundUdpProp extends BaseUdpProp {
}
定义用于配置UDP组播出站参数的类型
@Component
@ConfigurationProperties(prefix = "custom.udp.multicast.outbound")
public class MulticastOutboundUdpProp extends BaseUdpProp {
}
自定义的命令
public class ClusterCommand {
public static final String FINDING_COMMAND = "Finding cluster";
public static final String JOINING_COMMAND = "Joining cluster";
public static final String RESET_COMMAND = "Reset node";
}
配置UDP单播入站
配置了Message Router和Transformer消息组件,用来实现按内容进行不同通道分发的业务逻辑
@Configuration
public class UnicastInboundUdpConfig {
@Resource
private UnicastInboundUdpProp prop;
@Resource
private ObjectMapper objectMapper;
/**
* 用于配置UDP单播的适配器,对应前文中提到的Channel Adapter的概念。
* 该实例创建后,会自动产生对于目标端口的监听线程
* 这里我们只定义了通道的名称,没有创建通道,所以会使用默认的DirectChannel,
* 如果希望自定义通道,则可以调用adapter.setOutputChannel方法
*
* @author huangji
*/
@Bean
public UnicastReceivingChannelAdapter unicastAdapter() {
UnicastReceivingChannelAdapter adapter = new UnicastReceivingChannelAdapter(prop.getPort());
adapter.setOutputChannelName("UnicastInboundUdpChannel");
return adapter;
}
/**
* 填入inputChannel,用来接收从apdater输出的数据,
* 之后根据我们定义的判断逻辑,来分发给对应的目标通道
*
* @param payload 负载消息,会被通道自动注入到变量中
* @param headers 加了@Headers注解,会被通道自动注入消息头
* @return 返回通道名称,即最终分发的目标通道的名称
* @author huangji
*/
@Router(inputChannel = "UnicastInboundUdpChannel")
public String router(String payload, @Headers Map<String, Object> headers) {
String host = headers.get(IpHeaders.IP_ADDRESS).toString();
System.out.println("received unicast inbound from " + host);
System.out.println("content is " + payload);
try {
objectMapper.readValue(payload, new TypeReference<List<String>>() {
});
return "NodeNameListChannel";
} catch (JsonProcessingException ignored) {
}
try {
objectMapper.readValue(payload, NodeJoiningParameter.class);
return "JoiningParameterChannel";
} catch (JsonProcessingException ignored) {
}
if (payload.equals(ClusterCommand.RESET_COMMAND)) {
return "JoinedListeningChannel";
}
return "OtherChannel";
}
/**
* 这里用@Transformer注解定义了一个转换器组件,将负载类型转化成了目标类型
*
* @param payload 负载消息,会被通道自动注入到变量中
* @return 返回转化后的结果List<String>
* @throws JsonProcessingException
* @author huangji
*/
@Transformer(inputChannel = "NodeNameListChannel", outputChannel = "NodeNameTransformerChannel")
public List<String> nodeNameListTransformer(String payload) throws JsonProcessingException {
return objectMapper.readValue(payload, new TypeReference<>() {
});
}
/**
* 这里用@Transformer注解定义了一个转换器组件,将负载类型转化成了目标类型
*
* @param payload 负载消息,会被通道自动注入到变量中
* @return 返回转化后的结果
* @throws JsonProcessingException
* @author huangji
*/
@Transformer(inputChannel = "JoiningParameterChannel", outputChannel = "ParameterTransformerChannel")
public NodeJoiningParameter parameterTransformer(String payload) throws JsonProcessingException {
return objectMapper.readValue(payload, NodeJoiningParameter.class);
}
}
配置消息最终接收的通道
在这里进行自身业务逻辑的处理
@ServiceActivator(inputChannel = "NodeNameTransformerChannel")
public void nodeInfoListHandler(List<String> list, @Headers Map<String, Object> headers) {
...
}
@ServiceActivator(inputChannel = "ParameterTransformerChannel")
public void joiningParameterHandler(NodeJoiningParameter parameter, @Headers Map<String, Object> headers) {
String host = headers.get(IpHeaders.IP_ADDRESS).toString();
String udpHost = udpConfig.getMessageHandler().getHost();
if (host.equals(udpHost)) {
...
}
}
@ServiceActivator(inputChannel = "JoinedListeningChannel")
public void joinedListeningHandler(@Headers Map<String, Object> headers) {
String host = headers.get(IpHeaders.IP_ADDRESS).toString();
String udpHost = udpConfig.getMessageHandler().getHost();
if (host.equals(udpHost)) {
...
}
}
@ServiceActivator(inputChannel = "OtherChannel")
public void otherHandler(String payload) {
System.out.println("other payload: " + payload);
}
发送消息
有效负载的内容在代码中必须是String类型或者byte[]类型,否则会在发送消息时抛出异常
发送UDP广播消息
MulticastSendingMessageHandler multicastHandler= SpringUtil.getBean("multicastHandler");
multicastHandler.handleMessage(MessageBuilder.withPayload(ClusterCommand.FINDING_COMMAND).build());
发送UDP单播消息文章来源:https://www.toymoban.com/news/detail-792712.html
UnicastSendingMessageHandler messageHandler = outboundUdp.getMessageHandler();
messageHandler.handleMessage(MessageBuilder.withPayload(ClusterCommand.JOINING_COMMAND).build());
服务端
配置UDP组播入站
@Configuration
public class MulticastInboundUdpConfig {
@Value("${custom.udp.multicast.inbound.host}")
private String host;
@Value("${custom.udp.multicast.inbound.port}")
private Integer port;
/**
* 与单播的配置类似,这里也是配置组播的ChannelAdapter
*
*/
@Bean
public MulticastReceivingChannelAdapter multicastAdapter() {
MulticastReceivingChannelAdapter adapter = new MulticastReceivingChannelAdapter(host, port);
adapter.setOutputChannelName("MulticastInboundUdpChannel");
return adapter;
}
/**
* 这里也是配置消息的Router组件
*
*/
@Router(inputChannel = "MulticastInboundUdpChannel")
public String router(String payload, @Headers Map<String, Object> headers) {
String remoteHost = headers.get(IpHeaders.IP_ADDRESS).toString();
System.out.println("received multicast inbound from " + remoteHost);
System.out.println("content is " + payload);
if (payload.equals(ClusterCommand.FINDING_COMMAND)) {
return "FindingChannel";
}
return "OtherChannel";
}
}
配置UDP单播入站
@Configuration
public class UnicastInboundUdpConfig {
@Value("${custom.udp.unicast.inbound.port}")
private Integer port;
@Bean
public UnicastReceivingChannelAdapter unicastAdapter() {
UnicastReceivingChannelAdapter adapter = new UnicastReceivingChannelAdapter(port);
adapter.setOutputChannelName("UnicastInboundUdpChannel");
return adapter;
}
@Router(inputChannel = "UnicastInboundUdpChannel")
public String router(String payload, @Headers Map<String, Object> headers) {
String host = headers.get(IpHeaders.IP_ADDRESS).toString();
System.out.println("received unicast inbound from " + host);
if (payload.equals(ClusterCommand.JOINING_COMMAND)) {
return "JoiningChannel";
}
return "OtherChannel";
}
}
配置消息最终接收的通道
@ServiceActivator(inputChannel = "FindingChannel")
public void listenFindingHandler(@Headers Map<String, Object> headers) throws JsonProcessingException {
String host = headers.get(IpHeaders.IP_ADDRESS).toString();
UnicastSendingMessageHandler messageHandler = new UnicastSendingMessageHandler(host, udpOutboundPort);
String payload = objectMapper.writeValueAsString(fetchNodeNameList());
messageHandler.handleMessage(MessageBuilder.withPayload(payload).build());
}
@ServiceActivator(inputChannel = "JoiningChannel")
public void listenJoiningHandler(@Headers Map<String, Object> headers) throws IOException, InterruptedException {
String host = headers.get(IpHeaders.IP_ADDRESS).toString();
UnicastSendingMessageHandler messageHandler = new UnicastSendingMessageHandler(host, udpOutboundPort);
...
}
@ServiceActivator(inputChannel = "OtherChannel")
public void otherHandler(String payload) {
System.out.println("other payload: " + payload);
}
附录
消息头
Spring Integration Ip包含以下的头部信息,可以根据需要去获取自己想要的头部信息。文章来源地址https://www.toymoban.com/news/detail-792712.html
Header Name | IpHeaders Constant | Description |
---|---|---|
ip_hostname |
HOSTNAME |
The host name from which a TCP message or UDP packet was received. If lookupHost is false , this contains the IP address. |
ip_address |
IP_ADDRESS |
The IP address from which a TCP message or UDP packet was received. |
ip_port |
PORT |
The remote port for a UDP packet. |
ip_localInetAddress |
IP_LOCAL_ADDRESS |
The local InetAddress to which the socket is connected (since version 4.2.5). |
ip_ackTo |
ACKADDRESS |
The remote IP address to which UDP application-level acknowledgments are sent. The framework includes acknowledgment information in the data packet. |
ip_ackId |
ACK_ID |
A correlation ID for UDP application-level acknowledgments. The framework includes acknowledgment information in the data packet. |
ip_tcp_remotePort |
REMOTE_PORT |
The remote port for a TCP connection. |
ip_connectionId |
CONNECTION_ID |
A unique identifier for a TCP connection. Set by the framework for inbound messages. When sending to a server-side inbound channel adapter or replying to an inbound gateway, this header is required so that the endpoint can determine the connection to which to send the message. |
ip_actualConnectionId |
ACTUAL_CONNECTION_ID |
For information only. When using a cached or failover client connection factory, it contains the actual underlying connection ID. |
contentType |
MessageHeaders. CONTENT_TYPE
|
An optional content type for inbound messages Described after this table. Note that, unlike the other header constants, this constant is in the MessageHeaders class, not the IpHeaders class. |
到了这里,关于Spring Integration Ip 一个好用的TCP/UDP开发框架的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!