使用自定义MessageConverter创建消息时出现StreamBridge问题

Question

我们正在将我们的事件系统迁移到函数式编程模型。我们遵循了下一个“指南”，它对消费者非常有效，但使用StreamBridge的生产者没有正确创建消息。

我们有下一个错误：

java.lang.ClassCastException: class com.streamdemo.domain.Event cannot be cast to class java.lang.String (com.streamdemo.domain.Event is in unnamed module of loader 'app'; java.lang.String is in module java.base of loader 'bootstrap')

我们正在使用< code > 2021 . 0 . 0 spring-cloud版本。

通过简单的配置:

spring.cloud.stream:
  bindings:
    demo-channel-out-0:
      producer:
        use-native-encoding: true
      content-type: application/json

这是我们的自定义MessageConver，它用于我们所有的微服务，因此，保持消息的格式很重要：

public class StreamMessageConverter extends AbstractMessageConverter {

  private static final Logger log = LoggerFactory.getLogger(StreamMessageConverter.class);
  private static final MimeType JSON_MIME_TYPE = MimeType.valueOf("application/json");
  private final Class<?> clazz;
  private final ObjectMapper objectMapper;

  public StreamMessageConverter(ObjectMapper objectMapper) {
    this(objectMapper, Object.class);
  }

  public StreamMessageConverter(ObjectMapper objectMapper, Class<?> clazz) {
    super(JSON_MIME_TYPE);
    this.objectMapper = objectMapper;
    this.clazz = clazz;
  }

  protected boolean supports(Class<?> clazz) {
    return clazz.equals(clazz);
  }

  @Nullable
  protected Object convertFromInternal(Message<?> message, Class<?> targetClass, @Nullable Object conversionHint) {
    Object payload = message.getPayload();

    try {
      return payload instanceof byte[] ? this.objectMapper.readValue((byte[]) payload, targetClass)
          : this.objectMapper.readValue((String) payload, targetClass);
    } catch (IOException var6) {
      log.info("Unable to read json payload as object", var6);
      return null;
    }
  }

  @Nullable
  protected Object convertToInternal(Object payload, @Nullable MessageHeaders headers, @Nullable Object conversionHint) {
    try {
      return this.objectMapper.writeValueAsString(payload).getBytes("UTF-8");
    } catch (IOException var5) {
      log.info("Unable to write payload as json", var5);
      return null;
    }
  }
}

我们还使用< code>StreamBridge来生成事件:

@Service
public class MessageBroker {

  private final StreamBridge source;

  public MessageBroker(StreamBridge source) {
    this.source = source;
  }

  public void send(Event event) {
    source.send("demo-channel-out-0", event);
  }
}

当＜code＞源代码。send方法被调用，它在内部调用StreamMessageConverter。convertFromInternal当我们认为它不应该这样做时。在以前的模型（@EnableBinding，@StreamListener）中，只有StreamMessageConverter。convertToInternal被调用，以便将消息序列化并发送到Kafka主题。

我们创建了这个演示项目，您可以在其中重现错误。

Answer 1

所以发生的事情是正确的。StreamBridge依赖MessageCon的原因是因为我们定义了一个即时传递函数（函数

您的MessageConver似乎是在考虑某些期望的情况下编写的，这是不正确的，也从来没有正确过。您假设如果有效负载不是byte[]它是String。即使使用较旧的编程模型，也有有效负载可能是这两种类型以外的类型的情况。因此，我建议将以下代码块修复为正确的IF语句

return payload instanceof byte[] ? this.objectMapper.readValue((byte[]) payload, targetClass)
          : this.objectMapper.readValue((String) payload, targetClass);

。。。以确保它能处理另一种情况。

也就是说，你的转换器没有做任何特别的事情，特别是在当前的编程模型中。所以从技术上来说，要解决你的问题，你只需要将其全部删除。