springboot搭建非阻塞式tcp服务器和客户端
工作区间新建项目
D:\workspace\eshore-build\eshore-build-tcpnio
新建Application.java
D:\workspace\eshore-build\eshore-build-tcpnio\src\main\java\com\eshore\build\tcpnio\Application.java
其内容为
package com.eshore.build.tcpnio;
import java.text.SimpleDateFormat;
import java.util.Map;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.integration.annotation.MessagingGateway;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.annotation.Transformer;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.FixedSubscriberChannel;
import org.springframework.integration.ip.tcp.TcpInboundGateway;
import org.springframework.integration.ip.tcp.TcpReceivingChannelAdapter;
import org.springframework.integration.ip.tcp.TcpSendingMessageHandler;
import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory;
import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory;
import org.springframework.integration.ip.tcp.connection.TcpNioClientConnectionFactory;
import org.springframework.integration.ip.tcp.connection.TcpNioServerConnectionFactory;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import com.fasterxml.jackson.databind.ObjectMapper;
/**
* @author jiangyl
*
*/
@SpringBootApplication
@EnableScheduling
public class Application {
public static void main(String[] args) {
SpringApplication.run(Application.class, args);
}
@Autowired
private Gateway gateway;
@Scheduled(fixedDelay = 1000L)
public void sendMessageJob() {
String currentTime = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(System.currentTimeMillis());
String msg = "{\"currentTime\":\"" + currentTime + "\"}";
System.out.println("客户端发送消息" + msg);
gateway.sendMessage(msg);
}
// server
@Bean
public AbstractServerConnectionFactory serverConnectionFactory() {
return new TcpNioServerConnectionFactory(5678);
}
@Bean
public MessageChannel requestChannel() {
return new DirectChannel();
}
@Bean
public TcpInboundGateway tcpInboundGateway() {
TcpInboundGateway tcpInboundGateway = new TcpInboundGateway();
tcpInboundGateway.setConnectionFactory(serverConnectionFactory());
tcpInboundGateway.setRequestChannel(requestChannel());
return tcpInboundGateway;
}
@Transformer(inputChannel = "requestChannel", outputChannel = "requestChannel2")
public String serverConvert(byte[] bytes) {
return new String(bytes);
}
@ServiceActivator(inputChannel = "requestChannel2")
public String handleRequest(String msg) throws Exception {
System.out.println("服务端处理请求消息=" + msg);// server handle
ObjectMapper objectMapper = new ObjectMapper();
Map map = objectMapper.readValue(msg, Map.class);
map.put("result", "is processed");
return objectMapper.writeValueAsString(map);
}
// client
@Bean
public AbstractClientConnectionFactory clientConnectionFactory() {
return new TcpNioClientConnectionFactory("localhost", 5678);
}
@Component
@MessagingGateway(defaultRequestChannel = "sendMessageChannel")
public interface Gateway {
void sendMessage(String message);
}
@Bean
@ServiceActivator(inputChannel = "sendMessageChannel")
public TcpSendingMessageHandler tcpSendingMessageHandler() {
TcpSendingMessageHandler tcpSendingMessageHandler = new TcpSendingMessageHandler();
tcpSendingMessageHandler.setConnectionFactory(clientConnectionFactory());
return tcpSendingMessageHandler;
}
@Bean
public TcpReceivingChannelAdapter tcpReceivingChannelAdapter() {
TcpReceivingChannelAdapter tcpReceivingChannelAdapter = new TcpReceivingChannelAdapter();
tcpReceivingChannelAdapter.setConnectionFactory(clientConnectionFactory());
tcpReceivingChannelAdapter.setOutputChannelName("outputChannel");
return tcpReceivingChannelAdapter;
}
@Transformer(inputChannel = "outputChannel", outputChannel = "outputChannel2")
public String clientConvert(byte[] bytes) {
return new String(bytes);
}
/*@Bean
public MessageChannel outputChannel2() {
return new FixedSubscriberChannel(msg -> {
System.out.println("客户端处理响应消息=" + msg.getPayload());// client handle
});
}
@Bean
@ServiceActivator(inputChannel = "outputChannel2")
public MessageHandler handleResponse() {
return msg -> {
System.out.println("客户端处理响应消息=" + msg.getPayload());// client handle
};
}*/
@ServiceActivator(inputChannel = "outputChannel2")
public String handleResponse(String msg) throws Exception {
System.out.println("客户端处理响应消息=" + msg);// client handle
return null;
}
}
新建配置文件pom.xml
D:\workspace\eshore-build\eshore-build-tcpnio\pom.xml
其内容为
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.eshore.build</groupId>
<artifactId>eshore-build-tcpnio</artifactId>
<version>1.0.0</version>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.1.4.RELEASE</version>
</parent>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-json</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-integration</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-ip</artifactId>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<source>1.8</source>
<target>1.8</target>
</configuration>
</plugin>
</plugins>
</build>
</project>
执行
命令行切换到D:\workspace\eshore-build\eshore-build-tcpnio,执行命令
mvn spring-boot:run