仅拦截来自消费者的Spring云流消息

Question

我目前正在使用带有GlobalChannelInterceptor的Kafka绑定器的Spring Cloud Stream为我的Spring Boot微服务执行消息记录。

我有：

1. 生产者将消息发布到订阅频道

在消息从生产者发布到流并被消费者收听的整个过程中，可以观察到preSend方法被触发了两次：

1. 一次在生产者端-消息发布到流时

然而，出于日志记录的目的，我只需要在消费者端截获并记录消息。

是否有任何方法可以仅在一侧（例如消费者侧）截获SCS消息？

我将不胜感激对此事的任何想法。谢谢！

参考：

1. GlobalChannelInterceptor文档-https://docs.spring.io/spring-integration/api/org/springframework/integration/config/GlobalChannelInterceptor.html

编辑

制作人

public void sendToPushStream(PushStreamMessage message) {
        try {
            boolean results = streamChannel.pushStream().send(MessageBuilder.withPayload(new ObjectMapper().writeValueAsString(message)).build());
        log.info("Push stream message {} sent to {}.", results ? "successfully" : "not", StreamChannel.PUSH_STREAM);
        } catch (JsonProcessingException ex) {
            log.error("Unable to parse push stream message.", ex);
        }
    }

制片人的streamChannel

public interface StreamChannel {

    String PUSH_STREAM = "PushStream";

    @Output(StreamChannel.PUSH_STREAM)
    SubscribableChannel pushStream();

}

消费者

@StreamListener(StreamChannel.PUSH_STREAM)
public void handle(Message<PushStreamMessage> message) {
    log.info("Incoming stream message from {}, {}", streamChannel.pushStream(), message);

}

消费者的streamChannel

public interface StreamChannel {

    String PUSH_STREAM = "PushStream";

    @Input(StreamChannel.PUSH_STREAM)
    SubscribableChannel pushStream();

}

拦截器（公共库）

public class GlobalStreamInterceptor extends ChannelInterceptorAdapter {

    @Override
    public Message<?> preSend(Message<?> msg, MessageChannel mc) {
       log.info("presend " + msg);
        return msg;
    }

    @Override
    public void postSend(Message<?> msg, MessageChannel mc, boolean sent) {
        log.info("postSend " + msg);
    }

}

Answer 1

对，为什么不遵循GlobalChannelInterceptor选项并且不应用

通道名称将与之匹配的简单模式数组。

？

因此，您可能会遇到这样的情况：

@GlobalChannelInterceptor(patterns = Processor.INPUT)

或者为您的SCSt应用程序使用输入频道的自定义名称。