如何配置 tcp 服务器使用 spring 启动从多个客户端接收数据？

Question

我想配置 TCP 服务器以接收和回复来自多个客户端的数据。我搜索了许多其他线程，但找不到确切的方法。我是第一次使用 spring 集成，没有经验。

服务器要求

应该能够接收和回复特定客户端的数据（可以有多个客户端，每个客户端应该单独处理）
应该能够向客户端发送数据并等待特定超时的响应。
应该能够检测到客户端是否断开连接。如果客户端断开连接，则应关闭连接以节省内存。（在没有 spring 集成的早期方法中，我能够通过 ping 客户端来完成，并查看发送是否失败但不知道如何使用 spring 集成）

我尝试了下面的代码，其中我能够向客户端发送数据但可以实现我的上述要求

TCP 服务器配置：

@Configuration
public class TcpServerConfig {

    private List<TcpConnectionOpenEvent> clientList = new ArrayList<>();

    public List<TcpConnectionOpenEvent> getClientList() {
        return clientList;
    }

    @Bean
    public TcpReceivingChannelAdapter server(TcpNetServerConnectionFactory cf) {
        TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();
        adapter.setConnectionFactory(cf);
        adapter.setOutputChannel(inputChannel());
        return adapter;
    }

    @Bean
    public MessageChannel inputChannel() {
        return new QueueChannel();
    }

    @Bean
    public MessageChannel outputChannel() {
        return new DirectChannel();
    }

    @Bean
    public TcpNetServerConnectionFactory cf() {
        return new TcpNetServerConnectionFactory(1001);
    }

    @Bean
    public IntegrationFlow outbound() {
        return IntegrationFlows.from(outputChannel())
                .handle(sender())
                .get();
    }

    @Bean
    public MessageHandler sender() {
        TcpSendingMessageHandler tcpSendingMessageHandler = new TcpSendingMessageHandler();
        tcpSendingMessageHandler.setConnectionFactory(cf());
        return tcpSendingMessageHandler;
    }

    @Bean
    public ApplicationListener<TcpConnectionOpenEvent> listener() {
        return new ApplicationListener<TcpConnectionOpenEvent>() {

            @Override
            public void onApplicationEvent(TcpConnectionOpenEvent event) {
                outputChannel().send(MessageBuilder.withPayload("foo")
                        .setHeader(IpHeaders.CONNECTION_ID, event.getConnectionId())
                        .build());

                clientList.add(event);
            }
        };
    }
}

测试代码：

@Service
public class Test {

    private static final Logger LOGGER = LoggerFactory.getLogger(MessageServiceImpl.class);

    @Autowired
    TcpServerConfig tcpServerConfig;

    @Autowired
    private MessageChannel outputChannel;

    @Autowired
    private MessageChannel inputChannel;

    @Scheduled(fixedRate = 1000)
    void task() {
            LOGGER.info("Client count: " + tcpServerConfig.getClientList().size());

            for (TcpConnectionOpenEvent client : tcpServerConfig.getClientList()) {
                outputChannel.send(MessageBuilder.withPayload("foo")
                        .setHeader(IpHeaders.CONNECTION_ID, client.getConnectionId())
                        .build());
            }
        }
}

如有任何帮助，我们将不胜感激。

Answer 1

这是一种解决方案：

@SpringBootApplication
@EnableScheduling
public class So62877512ServerApplication {

    public static void main(String[] args) {
        SpringApplication.run(So62877512ServerApplication.class, args);
    }

    @Bean
    public IntegrationFlow serverIn(Handler handler) {
        return IntegrationFlows.from(Tcp.inboundAdapter(server()))
                .transform(Transformers.objectToString())
                .filter(handler, "existingConnection", spec -> spec
                        .discardFlow(f -> f
                                .handle(handler, "sendInitialReply")))
                .handle(handler, "reply")
                .get();
    }

    @Bean
    public IntegrationFlow serverOut() {
        return f -> f.handle(Tcp.outboundAdapter(server()));
    }

    @Bean
    public TcpServerConnectionFactorySpec server() {
        return Tcp.netServer(1234)
                .serializer(TcpCodecs.lf())
                .deserializer(TcpCodecs.lf()); // compatible with netcat
    }

}

@Component
@DependsOn("serverOut")
class Handler {

    private static final Logger LOG = LoggerFactory.getLogger(Handler.class);

    private final ConcurrentMap<String, BlockingQueue<Message<?>>> clients = new ConcurrentHashMap<>();

    private final MessageChannel out;

    private final TcpNetServerConnectionFactory server;

    public Handler(@Qualifier("serverOut.input") MessageChannel out, TcpNetServerConnectionFactory server) {
        this.out = out;
        this.server = server;
    }

    public boolean existingConnection(Message<?> message) {
        String connectionId = message.getHeaders().get(IpHeaders.CONNECTION_ID, String.class);
        boolean containsKey = this.clients.containsKey(connectionId);
        if (!containsKey) {
            this.clients.put(connectionId, new LinkedBlockingQueue<Message<?>>());
        }
        return containsKey;
    }

    public void sendInitialReply(Message<String> message) {
        LOG.info("Replying to " + message.getPayload());
        this.out.send(MessageBuilder.withPayload(message.getPayload().toUpperCase())
                .copyHeaders(message.getHeaders()).build());
    }

    @Scheduled(fixedDelay = 5000)
    public void sender() {
        this.clients.forEach((key, queue) -> {
            try {
                this.out.send(MessageBuilder.withPayload("foo")
                        .setHeader(IpHeaders.CONNECTION_ID, key).build());
                Message<?> reply = queue.poll(10, TimeUnit.SECONDS);
                if (reply == null) {
                    LOG.error("Timeout waiting for " + key);
                    this.server.closeConnection(key);
                }
                else {
                    LOG.info("Reply " + reply.getPayload() + " from " + key);
                }
            }
            catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                LOG.error("Interrupted");
            }
            catch (Exception e) {
                LOG.error("Failed to send to " + key, e);
            }
        });
    }

    public void reply(Message<String> in) {
        BlockingQueue<Message<?>> queue = this.clients.get(in.getHeaders().get(IpHeaders.CONNECTION_ID, String.class));
        if (queue != null) {
            queue.add(in);
        }
    }

    @EventListener
    public void closed(TcpConnectionCloseEvent event) {
        this.clients.remove(event.getConnectionId());
        LOG.info(event.getConnectionId() + " closed");
    }

}

$ nc localhost 1234
foo <- typed
FOO
foo
bar <- typed
foo
bar <- typed
foo

$ <- closed by server - timeout

2020-07-14 14:41:04.906  INFO 64763 --- [pool-1-thread-2] com.example.demo.Handler                 : Replying to foo
2020-07-14 14:41:13.841  INFO 64763 --- [   scheduling-1] com.example.demo.Handler                 : Reply bar from localhost:65115:1234:a9fc7e3d-4dda-4627-b765-4f0bb0835153
2020-07-14 14:41:21.465  INFO 64763 --- [   scheduling-1] com.example.demo.Handler                 : Reply bar from localhost:65115:1234:a9fc7e3d-4dda-4627-b765-4f0bb0835153
2020-07-14 14:41:36.473 ERROR 64763 --- [   scheduling-1] com.example.demo.Handler                 : Timeout waiting for localhost:65115:1234:a9fc7e3d-4dda-4627-b765-4f0bb0835153
2020-07-14 14:41:36.474  INFO 64763 --- [   scheduling-1] com.example.demo.Handler                 : localhost:65115:1234:a9fc7e3d-4dda-4627-b765-4f0bb0835153 closed

如何配置 tcp 服务器使用 spring 启动从多个客户端接收数据？

How to config tcp server to receive data from multiple client using spring boot?

spring-integration

spring-integration-dsl