【发布时间】：2019-03-28 22:53:39
【问题描述】：
我正在创建一个消息处理器，它接收来自一个 kafka 主题的事件，对其进行处理并将结果转发到另一个主题。
我使用@SendTo 创建了一个@KafkaListener 方法，在我想控制传出消息的密钥生成之前，它可以正常工作。
documentation (2.2.4.RELEASE) 建议通过子类化 KafkaTemplate 创建一个 bean 并覆盖它的 send(String topic, String data) 方法。
不幸的是，这不起作用，因为在我的情况下没有调用该方法。另一方面，send(Message<?> message) 被调用，但这无济于事。经过短暂的调试后发现MessagingMessageListenerAdapter 在输入是org.springframework.messaging.Message 的实例并且结果不是List 的情况下调用此方法。不幸的是，RecordMessagingMessageListenerAdapter 总是将 input 转换为 Message。
我是否将这个注释组合用于不是 spring kafka 的作者意图的东西，这是一个错误还是文档错误？
此外，Spring Boot 自动配置只有在我不创建自己的 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);
}
};
}
*/
}
更新
以 send(Message) 结尾的堆栈跟踪
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)
RecordMessagingMessageListenerAdapter
这里将接收到的记录转换为 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);
}
}
MessagingMessageListenerAdapter
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。
【问题讨论】：