如何使用 QueueChannel 和 ServiceActivator 正确配置 TCP inboundAdapter

Question

我正在尝试配置一个 TCP 套接字，它以不同消息的格式接收 name,value 格式的数据。这些消息平均每秒到达，有时更快，有时更慢。

我能够设置一个工作配置，但我对 Spring Integration 中实际发生的事情缺乏基本的了解。

我的配置文件如下所示：

@Configuration
@EnableIntegration
public class TCPSocketServerConfig
{
    @Bean
    public IntegrationFlow server(
        final CSVProcessingService csvProcessingService,
        @Value("${tcp.socket.server.port}") final int port
    )
    {
        return IntegrationFlows.from(
            Tcp.inboundAdapter(
                Tcp.nioServer(port)
                   .deserializer(serializer())
                   .leaveOpen(true)
            )
               .autoStartup(true)
               .outputChannel(queueChannel())
        ).transform(new ObjectToStringTransformer())
         .handle(csvProcessingService)
         .get();
    }

    @Bean(name = PollerMetadata.DEFAULT_POLLER)
    public PollerMetadata defaultPoller()
    {
        return Pollers.fixedDelay(50, TimeUnit.MILLISECONDS).get();
    }

    @Bean
    public MessageChannel queueChannel()
    {
        return MessageChannels.queue("queue", 50).get();
    }

    @Bean
    public ByteArrayLfSerializer serializer()
    {
        final ByteArrayLfSerializer serializer = new ByteArrayLfSerializer();

        serializer.setMaxMessageSize(10240);

        return serializer;
    }
}

CSVProcessingService 看起来像这样（缩写）：

@Slf4j
@Service
public class CSVProcessingService
{
    @ServiceActivator
    public void process(final String message)
    {
        log.debug("DATA RECEIVED: \n" + message);
        final CsvMapper csvMapper = new CsvMapper();
        final CsvSchema csvSchema = csvMapper.schemaFor(CSVParameter.class);

        if (StringUtils.contains(message, StringUtils.LF))
        {
            processMultiLineInput(message, csvMapper, csvSchema);
        }
        else
        {
            processSingleLineInput(message, csvMapper, csvSchema);
        }
    }
}

我对此配置的目标如下：

• 在配置的端口上接收消息
• 承受更高的负载而不会丢失消息
• 反序列化消息
• 将它们放入队列通道
• （最好也记录错误）
• 队列通道每 50 毫秒轮询一次，来自队列通道的消息传递到ObjectToStringTransformer
• 在转换器之后，转换后的消息被传递给CSVProcessingService 进行进一步处理

我是否正确地实现了所有这些目标，还是因为我误解了 Spring Integration 而犯了错误？是否有可能以某种方式将Poller 和@ServiceActivator 结合起来？

此外，我无法想象我配置的 IntegrationFlow 是如何“流动”的，也许有人可以帮助我更好地理解这一点。

编辑：

在 Artems 发表评论后，我重新设计了我的配置。现在看起来像这样：

@Configuration
@EnableIntegration
public class TCPSocketServerConfig
{
    @Value("${tcp.socket.server.port}") int port;

    @Bean
    public IntegrationFlow server(
        final CSVProcessingService csvProcessingService
    )
    {
        return IntegrationFlows.from(
            Tcp.inboundAdapter(
                tcpNioServer()
            )
               .autoStartup(true)
               .errorChannel(errorChannel())
        )
         .transform(new ObjectToStringTransformer())
         .handle(csvProcessingService)
         .get();
    }

    @Bean
    public AbstractServerConnectionFactory tcpNioServer()
    {
        return Tcp.nioServer(port)
                  .deserializer(serializer())
                  .leaveOpen(true)
                  .taskExecutor(
                      new ThreadPoolExecutor(0, 20,
                                             30L, TimeUnit.SECONDS,
                                             new SynchronousQueue<>(),
                                             new DefaultThreadFactory("TCP-POOL"))
                  ).get();
    }

    @Bean
    public MessageChannel errorChannel()
    {
        return MessageChannels.direct("errors").get();
    }

    @Bean
    public IntegrationFlow errorHandling()
    {
        return IntegrationFlows.from(errorChannel()).log(LoggingHandler.Level.DEBUG).get();
    }

    @Bean
    public ByteArrayLfSerializer serializer()
    {
        final ByteArrayLfSerializer serializer = new ByteArrayLfSerializer();

        serializer.setMaxMessageSize(10240);

        return serializer;
    }
}

我还从CSVProcessingService#process 方法中删除了@ServiceActivator 注释。

Answer 1

不确定是什么让您感到困惑，但您的配置和逻辑看起来不错。

您可能会错过这样一个事实，即您不需要在两者之间使用QueueChannel，因为AbstractConnectionFactory.processNioSelections() 已经是多线程的，它会安排一个任务来从套接字读取消息。所以，你只需要为Tcp.nioServer()配置一个合适的Executor。尽管默认情况下它是Executors.newCachedThreadPool()。

另一方面，使用内存中的QueueChannel，您确实可能会丢失消息，因为它们已经从网络中读取。

当您使用 Java DSL 时，您应该考虑在端点上使用 poller() 选项。如果您有 inputChannel 属性，@Poller 将在 @ServiceActivator 上工作，但在 handle() 中使用相同的属性将覆盖 inputChannel，因此您的 @Poller 将不会被应用。不要混淆 Java DSL 和注解配置！

您的配置中的其他一切都很好。