java - TCP/IP服务器的spring boot处理

Question

必须实现一个服务器以通过以太网连接处理以下协议(protocol):

Establishing a connection
The client connects to the configured server via TCP / IP.
After the connection has been established, the client initially sends a heartbeat message to the
Server:

{
  "MessageID": "Heartbeat"
}

Response:
{
  "ResponseCode": "Ok"
}

Communication process
To maintain the connection, the client sends every 10 seconds when inactive
Heartbeat message.
Server and client must close the connection if they are not receiving a message for longer than 20 seconds.
An answer must be given within 5 seconds to request. 
If no response is received, the connection must also be closed.
The protocol does not contain numbering or any other form of identification.
Communication partner when sending the responses makes sure that they are in the same sequence.

Message structure:
The messages are embedded in an STX-ETX frame.
STX (0x02) message ETX (0x03)
An `escaping` of STX and ETX within the message is not necessary since it is in JSON format

Escape sequence are following:

JSON.stringify ({"a": "\ x02 \ x03 \ x10"}) → "{" a \ ": " \ u0002 \ u0003 \ u0010 \ "}"

不仅应该使用心跳消息。一个典型的消息应该是这样的:

{
  "MessageID": "CheckAccess"
  "Parameters": {
    "MediaType": "type",
    "MediaData": "data"
  }
} 

以及适当的回应:

{
  "ResponseCode":   "some-code",
  "DisplayMessage": "some-message",
  "SessionID":      "some-id"
}

它应该是一个多客户端服务器。并且协议(protocol)没有任何标识。
但是，我们必须至少识别客户端发送它的 IP 地址。
找不到有关如何将此类服务器添加到 Spring Boot 应用程序并在启动时启用并为其处理输入和输出逻辑的解决方案。
任何建议都受到高度赞赏。

解决方案
为 TCP 服务器配置以下内容:

@Slf4j
@Component
@RequiredArgsConstructor
public class TCPServer {
    private final InetSocketAddress hostAddress;
    private final ServerBootstrap serverBootstrap;

    private Channel serverChannel;

    @PostConstruct
    public void start() {
        try {
            ChannelFuture serverChannelFuture = serverBootstrap.bind(hostAddress).sync();
            log.info("Server is STARTED : port {}", hostAddress.getPort());

            serverChannel = serverChannelFuture.channel().closeFuture().sync().channel();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    @PreDestroy
    public void stop() {
        if (serverChannel != null) {
            serverChannel.close();
            serverChannel.parent().close();
        }
    }
}

@PostConstruct在应用程序启动期间启动服务器。
它的配置也是:

@Configuration
@RequiredArgsConstructor
@EnableConfigurationProperties(NettyProperties.class)
public class NettyConfiguration {

    private final LoggingHandler loggingHandler = new LoggingHandler(LogLevel.DEBUG);
    private final NettyProperties nettyProperties;

    @Bean(name = "serverBootstrap")
    public ServerBootstrap bootstrap(SimpleChannelInitializer initializer) {
        ServerBootstrap bootstrap = new ServerBootstrap();
        bootstrap.group(bossGroup(), workerGroup())
                .channel(NioServerSocketChannel.class)
                .handler(loggingHandler)
                .childHandler(initializer);
        bootstrap.option(ChannelOption.SO_BACKLOG, nettyProperties.getBacklog());
        bootstrap.childOption(ChannelOption.SO_KEEPALIVE, nettyProperties.isKeepAlive());
        return bootstrap;
    }

    @Bean(destroyMethod = "shutdownGracefully")
    public NioEventLoopGroup bossGroup() {
        return new NioEventLoopGroup(nettyProperties.getBossCount());
    }

    @Bean(destroyMethod = "shutdownGracefully")
    public NioEventLoopGroup workerGroup() {
        return new NioEventLoopGroup(nettyProperties.getWorkerCount());
    }

    @Bean
    @SneakyThrows
    public InetSocketAddress tcpSocketAddress() {
        return new InetSocketAddress(nettyProperties.getTcpPort());
    }
}

初始化逻辑:

@Component
@RequiredArgsConstructor
public class SimpleChannelInitializer extends ChannelInitializer<SocketChannel> {

    private final StringEncoder stringEncoder = new StringEncoder();
    private final StringDecoder stringDecoder = new StringDecoder();

    private final QrReaderProcessingHandler readerServerHandler;
    private final NettyProperties nettyProperties;

    @Override
    protected void initChannel(SocketChannel socketChannel) {
        ChannelPipeline pipeline = socketChannel.pipeline();

        pipeline.addLast(new DelimiterBasedFrameDecoder(1024 * 1024, Delimiters.lineDelimiter()));
        pipeline.addLast(new ReadTimeoutHandler(nettyProperties.getClientTimeout()));
        pipeline.addLast(stringDecoder);
        pipeline.addLast(stringEncoder);
        pipeline.addLast(readerServerHandler);
    }
}

属性配置:

@Getter
@Setter
@ConfigurationProperties(prefix = "netty")
public class NettyProperties {
    @NotNull
    @Size(min = 1000, max = 65535)
    private int tcpPort;

    @Min(1)
    @NotNull
    private int bossCount;

    @Min(2)
    @NotNull
    private int workerCount;

    @NotNull
    private boolean keepAlive;

    @NotNull
    private int backlog;

    @NotNull
    private int clientTimeout;
}

以及来自 application.yml 的片段:

netty:
  tcp-port: 9090
  boss-count: 1
  worker-count: 14
  keep-alive: true
  backlog: 128
  client-timeout: 20

处理程序非常琐碎。
通过在控制台运行本地检查:

telnet localhost 9090

它在那里工作得很好。我希望它对客户的访问会很好。

Answer 1

由于该协议(protocol)不是基于 HTTP 之上的(不像 WebSocket 首先搭载 HTTP)，你唯一的选择是 自己使用 TCP 服务器 并在 Spring 环境中将其连接起来，以充分利用 Spring 。
网络 以低级 TCP/IP 通信而闻名，它很容易在 spring 应用程序中封装 Netty 服务器。
其实spring boot提供Netty HTTP server开箱即用，但 这不是你需要的 .
TCP communication server with Netty And SpringBoot项目是您需要的一个简单而有效的示例。
看看TCPServer这个项目使用 Netty 的 ServerBootstrap 来启动自定义 TCP 服务器。
拥有服务器后，您可以连接 Netty 编解码器 或 jackson 或您认为适合您的应用程序域数据的任何其他消息转换器编码/解码 .
[更新 - 2020 年 7 月 17 日]
针对对问题的更新理解(HTTP 和 TCP 请求都在同一端点上终止)，以下是更新的解决方案建议
----> HTTP 服务器 (be_http)
|
----> HAProxy -
|
----> TCP 服务器 (be_tcp)


要使此解决方案起作用，需要进行以下更改/添加:

在您现有的 Spring Boot 应用程序中添加基于 Netty 的监听器，或者为 TCP 服务器创建一个单独的 Spring Boot 应用程序。假设此端点正在监听端口 9090

上的 TCP 流量

添加 HAProxy 作为入口流量的终止端点

配置 HAProxy，使其将所有 HTTP 流量发送到端口 8080

上的现有 Spring Boot HTTP 端点(称为 be_http)

配置 HAProxy，以便将所有非 HTTP 流量发送到端口 9090 上的新 TCP spring boot 端点(称为 be_tcp)。

遵循 HAProxy 配置就足够了。这些是与此问题相关的摘录，请添加其他适用于正常 HAProxy 设置的 HAProxy 指令:

听443
模式 tcp
绑定(bind):443 名称 tcpsvr
/* 添加其他常规指令 */
tcp 请求检查延迟 1 秒
如果 HTTP 则接受 tcp 请求内容
如果 !HTTP 则接受 tcp 请求内容
如果是 HTTP，则使用服务器 be_http
如果 !HTTP 则使用服务器 be_tcp
/* 后端服务器定义 */
服务器 be_http 127.0.0.1:8080
服务器 be_tcp 127.0.0.1:9090 发送代理

以下 HAProxy 文档链接特别有用

Fetching samples from buffer contents - Layer 6

Pre-defined ACLs

tcp-request inspect-delay

tcp-request content

我个人会玩弄并验证 tcp 请求检查延迟 并根据实际需要对其进行调整，因为这有可能在最坏的情况下增加请求延迟，即已建立连接但尚无内容可用于评估请求是否为 HTTP。
满足 的需求我们必须至少识别客户端发送它的 IP 地址 ，您可以选择使用 Proxy Protocol同时将其发送回后端。我已经更新了上面的示例配置以在 be_tcp 中包含代理协议(protocol)(添加了 send_proxy)。我还从 be_http 中删除了 send_proxy，因为 spring boot 不需要它，相反，您可能会依赖常规的 X-Forwarded-For header 来代替 be_http 后端。
在 be_tcp 后端，你可以使用 Netty 的 HAProxyMessage使用 sourceAddress() 获取实际源 IP 地址API。总而言之，这是一个可行的解决方案。我自己使用了带有代理协议(protocol)的 HAProxy(在前端和后端)，它对于这项工作更加稳定。