Spring Integration队列满时如何每百条日志一次并清空队列？

我们通过@Gateway向队列发送消息。当队列已满时，不希望为每条消息记录日志，而是希望每100条消息记录一次日志并清空队列。请问在Spring Integration中是否有实现该需求的方案？

现有代码实现 #

// 向"INBOUND_CHANNEL"发送消息
@MessagingGateway(errorChannel = Constants.INBOUND_CHANNEL_ERR)
@Component
public interface CustomGateway {
    @Gateway(requestChannel = Constants.INBOUND_CHANNEL)
    public void sendMessage(Message<User> message);
}

// 将"INBOUND_CHANNEL"定义为QueueChannel
@Bean(name = Constants.INBOUND_CHANNEL)
public QueueChannel inboundFilePollingChannel() {
  return new QueueChannel(100);
}

// 将"INBOUND_CHANNEL_ERR"定义为QueueChannel
@Bean(name = Constants.INBOUND_CHANNEL_ERR)
public QueueChannel errorInboundChannel() {
    return new QueueChannel(100);
}

// "INBOUND_CHANNEL_ERR"对应的服务激活器
@ServiceActivator(inputChannel = Constants.INBOUND_CHANNEL_ERR)
public void handleErrors(org.springframework.messaging.Message<User> message) {
    // TODO：仅当"INBOUND_CHANNEL_ERR"队列已满时记录日志以进行限流并清空队列
    LOG.error("Dropping the message. ...");
}

实现方案 #

可以通过以下方式达成需求：

• 改用批量消费的错误处理逻辑
  抛弃单条消息处理的方式，配置轮询器实现批量拉取错误队列中的消息。当队列攒够100条消息时，一次性拉取所有消息，仅记录一次日志后完成处理（队列会被自动清空）。

• 修改服务激活器配置
  给@ServiceActivator添加poller属性，指定批量消费的参数，示例代码如下：

@ServiceActivator(inputChannel = Constants.INBOUND_CHANNEL_ERR, 
                  poller = @Poller(fixedDelay = "100", maxMessagesPerPoll = "100"))
public void handleErrors(List<Message<User>> messages) {
    if (!messages.isEmpty()) {
        LOG.error("批量丢弃{}条消息，原因：主队列INBOUND_CHANNEL已满", messages.size());
        // 批量拉取后队列已自动清空，无需额外操作
    }
}

• 适配队列容量与消费策略
  当前错误队列容量已设为100，正好匹配每100条记录一次日志的需求。如果需要更灵活的控制，可以增大错误队列容量，同时通过maxMessagesPerPoll参数限定每次处理的消息数量。

• 补充定时触发逻辑（可选）
  如果需要处理队列未满但长时间有积压消息的场景，可以在@Poller中配置定时触发规则，比如添加fixedRate或cron表达式，实现"攒够100条"或"到指定时间"任一条件满足时就触发批量处理。

原理说明：当主队列INBOUND_CHANNEL已满时，发送消息会抛出MessageDeliveryException，消息会被自动路由到错误通道INBOUND_CHANNEL_ERR。通过上述批量消费配置，错误队列会攒够100条消息后触发一次处理，仅记录一条批量日志，同时清空队列。

内容的提问来源于stack exchange，提问作者Raghavendra Prasad