Spring Cloud SteamKafka制作人留言

Question

我想建立一个弹簧-云-流-Kafka生产者与弹簧引导。

生产者正在工作，我可以使用来自kafka broker的消息，但这些消息还包含一些头部信息，如下所示：

contentType   "text/plain"originalContentType    "application/json;charset=UTF-8"{"message":"hello"}

我的POJO包含一个字段（字符串消息），所以我希望只有JSON字符串会发送给Kafka。

RestController中的方法test（）触发生产者：

@EnableBinding(ProducerChannels.class)
@SpringBootApplication
@RestController
public class KafkaStreamProducerApplication {

private MessageChannel consumer;

public KafkaStreamProducerApplication(ProducerChannels channels) {
    this.consumer = channels.consumer();
}

@PostMapping("/test/{message}")
public void test(@PathVariable String message) {
    Message<MyMessage> msg = MessageBuilder.withPayload(new MyMessage(message)).build();
    this.consumer.send(msg);
}

interface ProducerChannels {

    @Output
    MessageChannel consumer();
}

我的应用程序.属性

spring.cloud.stream.bindings.consumer.destination=consumer
spring.cloud.stream.bindings.consumer.content-type=application/json

如果你能推荐任何关于这个主题的文件或例子，我也将不胜感激。github上的示例通常非常简单，它们使用了大量的自动配置，并且没有任何解释。我使用的例子是RabbitMQ。

Answer 1

如果您想避免嵌入头（这样您就可以在一些非Spring云流应用程序中接收消息），请将生产者的headermode设置为raw。

请参阅生成器属性。

头模式