春季集成rabbit-mq json MessagingException

Question

我正在使用 Spring Integration 发送通知，作为一个错误测试用例，我正在发送格式错误的 JSON（地图）并收到 MessagingException，这似乎一直在继续……没有停止……我必须杀死应用程序。

所以想知道如何捕获这个，可能是通过errorChannel。代码示例会很有帮助。

我的 Spring 集成配置：

 <!-- channel to connect to disruption exchange -->
    <int-amqp:publish-subscribe-channel id="inputChannel"
                                        connection-factory="connectionFactory"
                                        exchange="notification.exchange"/>

    <int:json-to-object-transformer input-channel="inputChannel"
                                    output-channel="notificationChannel"
                                    type="java.util.Map"/>

    <int:channel id="notificationChannel">
        <int:interceptors>
            <int:wire-tap channel="loggingChannel"/>
        </int:interceptors>
    </int:channel>

    <int:logging-channel-adapter id="loggingChannel" log-full-message="true" logger-name="tapInbound" level="INFO"/>

    <!-- depending on the deviceType route to either apnsChannel or gcmChannel -->
    <int:router ref="notificationTypeRouter" input-channel="notificationChannel"/>

    <!-- apple push notification channel-->
    <int:channel id="apnsChannel"/>

    <!-- service activator to process disruptionNotificationChannel -->
    <int:service-activator input-channel="apnsChannel" ref="apnsPushNotificationService" method="pushNotification"/>

    <!-- google cloud messaging notification channel-->
    <int:channel id="gcmChannel"/>

    <!-- service activator to process disruptionNotificationChannel -->
    <int:service-activator input-channel="gcmChannel" ref="gcmPushNotificationService" method="pushNotification"/>

    <!-- error channel to may be log to file or email or store to db in the future -->
    <int:channel id="errorChannel"/>

    <int:service-activator input-channel="errorChannel" ref="notificationErrorHandler" method="handleFailedNotification"/>

    <!-- Infrastructure -->
    <rabbit:connection-factory id="connectionFactory"
                               host="${spring.rabbitmq.host}"
                               port="${spring.rabbitmq.port}"
                               username="${spring.rabbitmq.username}"
                               password="${spring.rabbitmq.password}"/>

    <rabbit:template id="amqpTemplate" connection-factory="connectionFactory"/>

    <rabbit:admin connection-factory="connectionFactory"/>

    <rabbit:fanout-exchange name="notification.exchange"/>

我还有一个错误处理程序：

public class NotificationErrorHandler {

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

    public void handleFailedNotification(Message<MessageHandlingException> message) {
        Map<String, Object> map = (Map) message.getPayload();
        Notification notification = Notification.fromMap(map);
        saveToBD(notification);
    }

    private void saveToBD(Notification notification) {
        LOG.error("[Notification-Error-Handler] Couldn't Send Push notification: device='{}', type='{}', pushId='{}', message='{}', uid='{}'",
                new Object[]{notification.getDevice(),
                        notification.getDeviceType(),
                        notification.getDeviceToken(),
                        notification.getBody(),
                        notification.getUid()});
    }
}

这是个例外：

Caused by: org.springframework.messaging.MessagingException: Failure occured in AMQP listener while attempting to convert and dispatch Message.; nested exception is org.springframework.integration.transformer.MessageTransformationException: failed to transform message; nested exception is com.fasterxml.jackson.core.JsonParseException: Unexpected character ('}' (code 125)): was expecting double-quote to start field name
 at [Source: [B@7a707c2c; line: 7, column: 2]
    at org.springframework.integration.amqp.channel.AbstractSubscribableAmqpChannel$DispatchingMessageListener.onMessage(AbstractSubscribableAmqpChannel.java:202)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:799)
    ... 10 common frames omitted
Caused by: org.springframework.integration.transformer.MessageTransformationException: failed to transform message; nested exception is com.fasterxml.jackson.core.JsonParseException: Unexpected character ('}' (code 125)): was expecting double-quote to start field name
 at [Source: [B@7a707c2c; line: 7, column: 2]
    at org.springframework.integration.transformer.AbstractTransformer.transform(AbstractTransformer.java:44)
    at org.springframework.integration.transformer.MessageTransformingHandler.handleRequestMessage(MessageTransformingHandler.java:68)
    at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:99)
    at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:78)
    at org.springframework.integration.dispatcher.BroadcastingDispatcher.invokeHandler(BroadcastingDispatcher.java:160)
    at org.springframework.integration.dispatcher.BroadcastingDispatcher.dispatch(BroadcastingDispatcher.java:142)
    at org.springframework.integration.amqp.channel.AbstractSubscribableAmqpChannel$DispatchingMessageListener.onMessage(AbstractSubscribableAmqpChannel.java:181)
    ... 11 common frames omitted
Caused by: com.fasterxml.jackson.core.JsonParseException: Unexpected character ('}' (code 125)): was expecting double-quote to start field name
 at [Source: [B@7a707c2c; line: 7, column: 2]
    at com.fasterxml.jackson.core.JsonParser._constructError(JsonParser.java:1419)
    at com.fasterxml.jackson.core.base.ParserMinimalBase._reportError(ParserMinimalBase.java:508)
    at com.fasterxml.jackson.core.base.ParserMinimalBase._reportUnexpectedChar(ParserMinimalBase.java:437)

希望有人能提供帮助。

提前致谢 通用汽车

---

根据@Gary 的回答所做的更改及其现在的工作：

<!-- Infrastructure -->
    <rabbit:connection-factory id="connectionFactory"
                               host="${spring.rabbitmq.host}"
                               port="${spring.rabbitmq.port}"
                               username="${spring.rabbitmq.username}"
                               password="${spring.rabbitmq.password}"/>

    <rabbit:admin connection-factory="connectionFactory"/>

    <rabbit:template id="amqpTemplate" connection-factory="connectionFactory"/>

    <rabbit:direct-exchange name="notification.direct">
        <rabbit:bindings>
            <rabbit:binding queue="notification.queue" key="notification.queue"/>
        </rabbit:bindings>
    </rabbit:direct-exchange>

    <rabbit:queue id="notification.queue" name="notification.queue"/>

    <int-amqp:inbound-channel-adapter channel="inputChannel"
                                      queue-names="notification.queue"
                                      connection-factory="connectionFactory"
                                      error-channel="errorChannel"/>

    <int:json-to-object-transformer input-channel="inputChannel"
                                    output-channel="notificationChannel"
                                    type="java.util.Map"/>

    <int:channel id="notificationChannel">
        <int:interceptors>
            <int:wire-tap channel="loggingChannel"/>
        </int:interceptors>
    </int:channel>

    <int:logging-channel-adapter id="loggingChannel" log-full-message="true" logger-name="tapInbound" level="INFO"/>

    <!-- depending on the deviceType route to either apnsChannel or gcmChannel -->
    <int:router ref="notificationTypeRouter" input-channel="notificationChannel"/>

    <!-- apple push notification channel-->
    <int:channel id="apnsChannel"/>

    <!-- service activator to process disruptionNotificationChannel -->
    <int:service-activator input-channel="apnsChannel" ref="apnsPushNotificationService" method="pushNotification"/>

    <!-- google cloud messaging notification channel-->
    <int:channel id="gcmChannel"/>

    <!-- service activator to process disruptionNotificationChannel -->
    <int:service-activator input-channel="gcmChannel" ref="gcmPushNotificationService" method="pushNotification"/>

    <!-- no op channel where message is logged for unknown devices -->
    <int:channel id="noOpChannel"/>

    <!-- service activator to process disruptionNotificationChannel -->
    <int:service-activator input-channel="noOpChannel" ref="noOpPushNotificationService" method="pushNotification"/>

    <!-- error channel to may be log to file or email or store to db in the future -->
    <int:channel id="errorChannel"/>

    <int:service-activator input-channel="errorChannel" ref="notificationErrorHandler"/>

Answer 1

为什么要使用发布-订阅频道开始流程？使用发布/订阅渠道进行消息分发是不正常的。

如果您可以改用消息驱动的通道适配器，则可以添加错误通道。

您不能将错误频道添加到发布-订阅频道。但是，您可以注入错误处理程序（实现 org.springframework.util.ErrorHandler）并在检测到致命错误时抛出 AmqpRejectAndDontRequeueException。

您也可以在通道中使用 Json MessageConverter，而不是在流下游使用 Json 转换器；在这种情况下，默认错误处理程序将检测到消息转换异常并拒绝该消息而不是重新排队。