在 Spring Integration 中使用 JSON 转换器答案

【问题标题】：Use JSON transformer in Spring Integration在 Spring Integration 中使用 JSON 转换器
【发布时间】：2019-05-16 13:38:16
【问题描述】：

我有一个问题，在我能找到的任何示例中似乎都没有解决。

我的应用程序读取 JSON 消息的 ActiveMQ 主题。它将基于此数据构建一个全新的出站 REST 调用。请注意，这不是“转换”。它被赋予“X”产生“Y”即ServiceActivator。

到目前为止我的流量是

public IntegrationFlow splitInputFlow() {
    return IntegrationFlows.from("inboundJmsChannel")
            .split()
            .log(LoggingHandler.Level.DEBUG)
            .route(Message.class, m -> m.getHeaders().get("x-bn-class").equals("Healthcheck.class") ? "healthcheckChannel" : "metricChannel")
            .get();
}

public IntegrationFlow healthcheckFlow() {
    return IntegrationFlows.from("healthcheckChannel")
            .log(LoggingHandler.Level.DEBUG)
        .transform(Transformers.fromJson(Healthcheck.class))
        .handle("healthcheckActivator", "process")
        .get();
}

有很多关于如何使用弹簧变压器的例子。我什至考虑过尝试使用 MessageConverter。但我不明白为什么它会有所帮助，而且这似乎不是正常的方法。

这里的主要问题是集成调用 healthcheckActivator.process(String payload)。有效负载本身是预期的有效 JSON 字符串。

我有点惊讶它没有调用 healthcheckActivator.process(Message payload) 但这无济于事，所以没关系。

真正的问题是它为什么不调用 healthcheckActivator.process(Healthcheck healthcheck)？

其实我明白“为什么”。这是因为 DSL 生成了一个内部通道来将这些步骤联系在一起，据我所知，通道上的任何内容都是 spring.messaging.Message。

进入 SA 后，我可以轻松地实例化我的 Healthcheck 对象。但这留下了一个棘手的问题：整个转换步骤有什么好处？如果它总是将对象“序列化”回 Message ——那有什么意义呢。

就像我说的，我认为我在这里遗漏了一些基本的东西。

编辑我的新想法（也可能是最后一个想法）可能是我发布错了。

要发布它，我正在使用

jmsTemplate.convertAndSend(topicName, healthcheck, messagePostProcessor -> {
            messagePostProcessor.setJMSType("TextMessage");
            messagePostProcessor.setStringProperty("x-bn-class", "Healthcheck.class");
            messagePostProcessor.setStringProperty("x-bn-service-name", restEndpoint.getServiceName());
            messagePostProcessor.setStringProperty("x-bn-service-endpoint-name", restEndpoint.getEndpointName());
            messagePostProcessor.setLongProperty("x-bn-heathcheck-timestamp", queryDate);
            messagePostProcessor.setStringProperty("x-bn-healthcheck-status", subsystemStatus.getStatus(subsystemStatus));
            messagePostProcessor.setIntProperty("httpStatus", httpStatus.value());
            return messagePostProcessor;
        });

SI process(String payload)方法中到达的是：

LoggingHandler - GenericMessage [payload={"healthcheckType":"LOCAL","outcome":"PASS","dependencyType":"DB","endpoint":"NODE TABLE","description":"Read from DB","durationSecs":0.025}, headers={x-bn-service-name=TG10-CS2, x-bn-service-endpoint-name=TG Q10-CS2 Ready Check, jms_destination=topic://HEALTH_MONITOR, _type=com.healthcheck.response.Healthcheck, x-bn-heathcheck-timestamp=1558356538000, priority=4, jms_timestamp=1558356544244, x-bn-healthcheck-status=SEV0, jms_redelivered=false, x-bn-class=Healthcheck.class, httpStatus=200, jms_type=TextMessage, id=b29ffea7-7128-c543-9a14-8bab450f0ac6, jms_messageId=ID:39479-1558356520091-1:2:1:1:1, timestamp=1558356544409}]

我之前没有注意到 jms_destination 标头中的 _type 参数。但在我开始搞砸之前（因为它不起作用），这是其他团队提供的正确类名。

我还没有实现 JMS 消息转换器。但是提供的 SimpleMessageConverter 似乎应该完全符合我的要求。

【问题讨论】：

标签： spring spring-integration

【解决方案1】：

你的理解是正确的；对我来说效果很好，所以正在发生其他事情......

@SpringBootApplication
public class So56169938Application {

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

    @Bean
    public IntegrationFlow flow() {
        return IntegrationFlows.from(() -> "{\"foo\":\"bar\"}", e -> e.poller(Pollers.fixedDelay(5000)))
            .transform(Transformers.fromJson(Foo.class))
            .handle("myBean", "method")
            .get();
    }

    @Bean
    public MyBean myBean() {
        return new MyBean();
    }

    public static class MyBean {

        public void method(Foo foo) {
            System.out.println(foo);
        }

    }

    public static class Foo {

        private String foo;

        String getFoo() {
            return this.foo;
        }

        void setFoo(String foo) {
            this.foo = foo;
        }

        @Override
        public String toString() {
            return "Foo [foo=" + this.foo + "]";
        }

    }

}

和

Foo [foo=bar]
Foo [foo=bar]
Foo [foo=bar]
Foo [foo=bar]
Foo [foo=bar]
Foo [foo=bar]

【讨论】：

感谢您的示例。我花了很长时间查看另一个团队给我的生成类。但是即使在我对其进行了重大更改之后，我也没有取得任何进展。但我不喜欢人们在我试图回答他们的问题时从不承认。所以这是我的正式感谢！
我会在有时间的时候再看看这个。但目前上述方法仍然失败，这在 SA Healthcheck 健康检查中完全符合预期； ObjectMapper 映射器 = 新 ObjectMapper(); healthcheck = mapper.readValue(payload, Healthcheck.class);很奇怪。
再考虑一下 Artem Bilan 的回应，也许我找错地方了。我添加了发布信息和队列中的内容的日志行。 @Gary Russell 现在这更有意义了吗？
否；现在它变得更没有意义了——SimpleMessageConverter 对 JSON 一无所知，所以如果你正在获取 JSON，就不能使用它。为什么不在JsonToObjectTransformer 中设置断点？
哇。这似乎是一个超级好主意！我扎根于实现并设置断点。他们没有一个被绊倒。所以看起来 SI 根本没有调用转换。我不会想到它甚至可以做到这一点。即使映射失败，我已经证明它没有，我希望它至少总是调用它。我无法摆脱 JMS 标头中缺少一些我不知道的东西的感觉。

【解决方案2】：

嗯，Spring Integration 是一个消息传递框架。它通过中间的通道将消息从端点传输到端点。这已经是目标端点以正确方式处理消费消息的责任。该框架不关心payload。它实际上是目标应用程序的业务部分。这就是我们如何使框架组件尽可能通用，从而为最终用户的目标业务类型留出空间。

无论如何，框架提供了一些与有效负载交互的机制。我们称之为POJO 方法调用。因此，您为某些业务提供任意合同，但遵循一些 Spring 集成规则：https://docs.spring.io/spring-integration/docs/current/reference/html/#service-activator。

所以，根据您的描述，它不适用于healtcheckActivator.process(Healthcheck healthcheck)，这真是令人惊讶。你的transform(Transformers.fromJson(Healthcheck.class)) 应该真正产生一个Message 和Healthcheck 对象作为有效负载。框架参考方法签名并尝试将有效负载和/或标头映射到方法调用参数，将整个消息作为数据容器委托给方法调用。

从这里很高兴看到您的 healtcheckActivator.process() 方法来确定为什么 transform(Transformers.fromJson(Healthcheck.class)) 结果 cannon 映射到该方法参数。

【讨论】：