Spring 集成从 RabbitMQ 迁移到 Redis 以共享应用程序事件

Question

我们正在微服务应用程序中从 RabbitMQ 迁移到 Redis。

这是我们的服务激活器

@ServiceActivator(inputChannel = ApplicationEventChannelNames.REMOTE_CHANNEL)
public void handleApplicationEvent(@Header(value = ApplicationEventHeaders.APPLICATION_EVENT) final ApplicationEvent event,
                                   @Payload Object message) {
...
}

最初我们遇到了一个问题，我们在SimpleMessageConverter 中丢失了应用程序事件。我们通过实现CustomRedisMessageConverter 并将应用程序事件放入fromMessage 方法中的有效负载中并从有效负载中检索它并在toMessage 方法中使用应用程序事件创建新的消息头来解决它。

    @Override
    public Object fromMessage(Message<?> message, Class<?> targetClass) {
        if (message.getHeaders().get(ApplicationEventHeaders.APPLICATION_EVENT) != null) {

            Map<String, Object> map = new HashMap<>();

            map.put("headers", ((ApplicationEvent) message.getHeaders().get(ApplicationEventHeaders.APPLICATION_EVENT)).getName());
            map.put("payload", message.getPayload());

            GenericMessage<Map<String, Object>> msg = new GenericMessage<>(map, message.getHeaders());

            return super.fromMessage(msg, targetClass);
        }
        return super.fromMessage(message, targetClass);
    }

    @Override
    public Message<?> toMessage(Object payload, MessageHeaders headers) {

        try {
            final Map<String, ?> message = new ObjectMapper().readValue((String) payload, new TypeReference<Map<String, ?>>() {});

            if (message.get("headers") != null) {
                final Map<String, Object> messageHeaders = new HashMap<>(headers);

                messageHeaders.put(ApplicationEventHeaders.APPLICATION_EVENT, new ApplicationEvent((String) message.get("headers")));

                return super.toMessage(message.get("payload"), new MessageHeaders(messageHeaders));
            }
        } catch (JsonProcessingException exception) {
            /* Intentionally left blank */
        }

        return super.toMessage(payload, headers);
    }

我们想知道是否有更好的方法来做到这一点？

最后，服务激活器中的有效负载以LinkedHashMap 的形式出现，但我们希望它是一个对象。使用 RabbitMQ 可以解决此问题。

有没有办法在 Redis 中做同样的事情？还是我们使用 headers 来跟踪有效载荷的类型并手动将它们转换为对象？

更新 - REDIS 配置

    @Bean
    public RedisInboundChannelAdapter applicationEventInboundChannelAdapter(@Value(value = "${com.xxx.xxx.xxx.integration.spring.topic}") String topic,
                                                                            MessageChannel applicationEventRemoteChannel,
                                                                            RedisConnectionFactory connectionFactory) {



        final RedisInboundChannelAdapter inboundChannelAdapter = new RedisInboundChannelAdapter(connectionFactory);
        inboundChannelAdapter.setTopics(topic);
        inboundChannelAdapter.setOutputChannel(applicationEventRemoteChannel);
        inboundChannelAdapter.setErrorChannel(errorChannel());
        inboundChannelAdapter.setMessageConverter(new CustomRedisMessageConverter());

        return inboundChannelAdapter;
    }

    @ServiceActivator(inputChannel = "errorChannel")
    public void processError(MessageHandlingException exception) {

        try {

            logger.error(
                    "Could not process {}, got exception: {}",
                    exception.getFailedMessage().getPayload(),
                    exception.getMessage());

            logger.error(
                    ExceptionUtils.readStackTrace(exception));

        } catch (Throwable throwable) {

            logger.error(
                    "Got {} during processing with message: {} ",
                    MessageHandlingException.class.getSimpleName(),
                    exception);
        }
    }

    @Bean
    @ServiceActivator(inputChannel = ApplicationEventChannelNames.LOCAL_CHANNEL)
    public RedisPublishingMessageHandler redisPublishingMessageHandler(@Value(value = "${com.xxx.xxx.xxx.integration.spring.topic}") String topic,
                                                                       RedisConnectionFactory redisConnectionFactory) {

        final RedisPublishingMessageHandler redisPublishingMessageHandler = new RedisPublishingMessageHandler(redisConnectionFactory);


        redisPublishingMessageHandler.setTopic(topic);
        redisPublishingMessageHandler.setSerializer(new Jackson2JsonRedisSerializer<>(String.class));
        redisPublishingMessageHandler.setMessageConverter(new CusomRedisMessageConverter());
        return redisPublishingMessageHandler;
    }

    /*
     * MessageChannel
     */

    @Bean
    public MessageChannel errorChannel() {
        return new DirectChannel();
    }

Answer 1

Redis 不支持标头，因此您必须将它们嵌入到正文中。请参阅EmbeddedJsonHeadersMessageMapper，它可以提供给双方的org.springframework.integration.support.converter.SimpleMessageConverter。