我正在创建一个消息处理器，它接收来自一个kafka主题的事件，处理它并将结果转发给另一个主题。
我用@KafkaListener创建了一个@SendTo方法，它可以正常工作，直到我想控制传出消息的密钥生成。
documentation (2.2.4.RELEASE)建议通过对KafkaTemplate进行子类化并覆盖它的send(String topic, String data)方法来创建一个bean。
不幸的是，这不起作用，因为在我的情况下不调用该方法。另一方面，send(Message<?> message)被召唤，但这没有帮助。经过短暂的调试后发现MessagingMessageListenerAdapter调用此方法，以防输入是org.springframework.messaging.Message的实例，结果不是List。不幸的是，RecordMessagingMessageListenerAdapter总是将输入转换为Message。我是否将这个注释组合用于spring kafka作者的意图，这是一个错误还是文档错误？
除此之外，只有当我不创建自己的KafkaTemplate bean时，弹簧启动自动配置才会起作用。如果我创建了被覆盖的模板，那么我必须自己创建KafkaListenerContainerFactory并设置回复模板以使@SendTo再次工作。
这是我的示例代码。尽可能简单。
@SpringBootApplication
@Slf4j
public class SpringKafkaExampleApplication {
public static void main(String[] args) {
SpringApplication.run(SpringKafkaExampleApplication.class, args);
}
@KafkaListener(topics = "${example.topics.input}")
@SendTo("${example.topics.output}")
public String process(final byte[] payload) {
String message = new String(payload, StandardCharsets.UTF_8);
log.info(message);
return message;
}
/*
//To set my custom KafkaTemplate as replyTemplate
@Bean
public ConcurrentKafkaListenerContainerFactory<String, byte[]> kafkaListenerContainerFactory(KafkaTemplate<String, String> kafkaTemplate,
ConsumerFactory<String, byte[]> consumerFactory) {
ConcurrentKafkaListenerContainerFactory<String, byte[]> factory =
new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory);
factory.setReplyTemplate(kafkaTemplate);
return factory;
}
//My KafkaTemplate with overridden send(topic, data) method
@Bean
public KafkaTemplate<String, String> kafkaTempate(ProducerFactory<String, String> producerFactory) {
return new KafkaTemplate<String, String>(producerFactory) {
@Override
public ListenableFuture<SendResult<String, String>> send(String topic, String data) {
return super.send(topic, "some_generated_key", data);
}
};
}
*/
}
UPDATE
send:215, KafkaTemplate (org.springframework.kafka.core)
sendReplyForMessageSource:449, MessagingMessageListenerAdapter (org.springframework.kafka.listener.adapter)
sendSingleResult:416, MessagingMessageListenerAdapter (org.springframework.kafka.listener.adapter)
sendResponse:402, MessagingMessageListenerAdapter (org.springframework.kafka.listener.adapter)
handleResult:324, MessagingMessageListenerAdapter (org.springframework.kafka.listener.adapter)
onMessage:81, RecordMessagingMessageListenerAdapter (org.springframework.kafka.listener.adapter)
onMessage:50, RecordMessagingMessageListenerAdapter (org.springframework.kafka.listener.adapter)
这里将接收到的记录转换为Message对象。
@Override
public void onMessage(ConsumerRecord<K, V> record, Acknowledgment acknowledgment, Consumer<?, ?> consumer) {
Message<?> message = toMessagingMessage(record, acknowledgment, consumer);
if (logger.isDebugEnabled()) {
logger.debug("Processing [" + message + "]");
}
try {
Object result = invokeHandler(record, acknowledgment, message, consumer);
if (result != null) {
handleResult(result, record, message);
}
}
String由KafkaListener方法返回，因此将调用sendSingleResult(result, topic, source)。
protected void sendResponse(Object result, String topic, @Nullable Object source, boolean messageReturnType) {
if (!messageReturnType && topic == null) {
if (this.logger.isDebugEnabled()) {
this.logger.debug("No replyTopic to handle the reply: " + result);
}
}
else if (result instanceof Message) {
this.replyTemplate.send((Message<?>) result);
}
else {
if (result instanceof Collection) {
((Collection<V>) result).forEach(v -> {
if (v instanceof Message) {
this.replyTemplate.send((Message<?>) v);
}
else {
this.replyTemplate.send(topic, v);
}
});
}
else {
sendSingleResult(result, topic, source);
}
}
}
private void sendSingleResult(Object result, String topic, @Nullable Object source) {
byte[] correlationId = null;
boolean sourceIsMessage = source instanceof Message;
if (sourceIsMessage
&& ((Message<?>) source).getHeaders().get(KafkaHeaders.CORRELATION_ID) != null) {
correlationId = ((Message<?>) source).getHeaders().get(KafkaHeaders.CORRELATION_ID, byte[].class);
}
if (sourceIsMessage) {
sendReplyForMessageSource(result, topic, source, correlationId);
}
else {
this.replyTemplate.send(topic, result);
}
}
@SuppressWarnings("unchecked")
private void sendReplyForMessageSource(Object result, String topic, Object source, byte[] correlationId) {
MessageBuilder<Object> builder = MessageBuilder.withPayload(result)
.setHeader(KafkaHeaders.TOPIC, topic);
if (this.replyHeadersConfigurer != null) {
Map<String, Object> headersToCopy = ((Message<?>) source).getHeaders().entrySet().stream()
.filter(e -> {
String key = e.getKey();
return !key.equals(MessageHeaders.ID) && !key.equals(MessageHeaders.TIMESTAMP)
&& !key.equals(KafkaHeaders.CORRELATION_ID)
&& !key.startsWith(KafkaHeaders.RECEIVED);
})
.filter(e -> this.replyHeadersConfigurer.shouldCopy(e.getKey(), e.getValue()))
.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
if (headersToCopy.size() > 0) {
builder.copyHeaders(headersToCopy);
}
headersToCopy = this.replyHeadersConfigurer.additionalHeaders();
if (!ObjectUtils.isEmpty(headersToCopy)) {
builder.copyHeaders(headersToCopy);
}
}
if (correlationId != null) {
builder.setHeader(KafkaHeaders.CORRELATION_ID, correlationId);
}
setPartition(builder, ((Message<?>) source));
this.replyTemplate.send(builder.build());
}
source现在是一条消息 - >将调用sendReplyForMessageSource。
经过简短的调试后发现，如果输入是
org.springframework.messaging.Message的实例，MessagingMessageListenerAdapter会调用此方法
这不正确;当侦听器方法返回Message<?>（或Collection<Message<?>>）时调用它。
码：
protected void sendResponse(Object result, String topic, @Nullable Object source, boolean messageReturnType) {
if (!messageReturnType && topic == null) {
if (this.logger.isDebugEnabled()) {
this.logger.debug("No replyTopic to handle the reply: " + result);
}
}
else if (result instanceof Message) {
this.replyTemplate.send((Message<?>) result);
}
else {
if (result instanceof Collection) {
((Collection<V>) result).forEach(v -> {
if (v instanceof Message) {
this.replyTemplate.send((Message<?>) v);
}
else {
this.replyTemplate.send(topic, v);
}
});
}
else {
sendSingleResult(result, topic, source);
}
}
}
自定义出站密钥的最简单方法是更改方法以返回Message<String>。从该文档链接向下滚动到...
如果侦听器方法返回Message或Collection>，则侦听器方法负责设置回复的邮件头。例如，在处理来自ReplyingKafkaTemplate的请求时，您可能会执行以下操作：
@KafkaListener(id = "messageReturned", topics = "someTopic")
public Message<?> listen(String in, @Header(KafkaHeaders.REPLY_TOPIC) byte[] replyTo,
@Header(KafkaHeaders.CORRELATION_ID) byte[] correlation) {
return MessageBuilder.withPayload(in.toUpperCase())
.setHeader(KafkaHeaders.TOPIC, replyTo)
.setHeader(KafkaHeaders.MESSAGE_KEY, 42)
.setHeader(KafkaHeaders.CORRELATION_ID, correlation)
.setHeader("someOtherHeader", "someValue")
.build();
}
除此之外，只有当我不创建自己的KafkaTemplate bean时，Spring boot auto配置才会起作用。
为了在回复模板中启动连线，必须将其声明为KafkaTemplate<Object, Object>。