在spring引导项目中使用kafka模板发送消息
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
@Service
@Slf4j
class KafkaMessagePublisherImpl {
@Autowired
private KafkaTemplate kafkaAsyncPublisher;
public void sendMessage() {
ListenableFuture listenableFuture = kafkaAsyncPublisher.send(
"test_topikc",
"key_1",
"{\"greeting\":\"Hello\"}");
listenableFuture.addCallback(new ListenableFutureCallback<SendResult<?, ?>>() {
@Override
public void onSuccess(final SendResult<?, ?> message) {
System.out.println("Sent");
}
@Override
public void onFailure(final Throwable throwable) {
System.out.println("Message sending failed");
}
});
}
public ProducerFactory<String, String> producerFactory() {
Map<String, Object> configProps = new HashMap<>();
configProps.put(
ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
"brokers1:9092,brokers2:9092");
configProps.put(
ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
StringSerializer.class);
configProps.put(
ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
StringSerializer.class);
return new DefaultKafkaProducerFactory<>(configProps);
}
@Bean(name = "kafkaAsyncPublisher")
public KafkaTemplate<String, String> kafkaAsyncPublisher() {
return new KafkaTemplate<>(producerFactory());
}
}
在使用 sendMessage
方法,出现以下错误
Caused by: org.springframework.kafka.core.KafkaProducerException: Failed to send; nested exception is org.apache.kafka.common.errors.TimeoutException: Failed to update metadata after 60000 ms.
at org.springframework.kafka.core.KafkaTemplate$1.onCompletion(KafkaTemplate.java:259)
at org.apache.kafka.clients.producer.KafkaProducer.doSend(KafkaProducer.java:760)
at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:701)
at org.springframework.kafka.core.DefaultKafkaProducerFactory$CloseSafeProducer.send(DefaultKafkaProducerFactory.java:170)
at org.springframework.kafka.core.KafkaTemplate.doSend(KafkaTemplate.java:245)
at org.springframework.kafka.core.KafkaTemplate.send(KafkaTemplate.java:157)
at com.tesco.fps.messaging.service.impl.KafkaMessagePublisherImpl.lambda$sendMessage$0(KafkaMessagePublisherImpl.java:52)
... 34 common frames omitted
Caused by: org.apache.kafka.common.errors.TimeoutException: Failed to update metadata after 60000 ms.
``` `Springboot:1.5.10.RELEASE` 以及 `compile('org.apache.kafka:kafka_2.12:0.11.0.0')` 作为渐变依赖项与 `spring-cloud-stream`
1条答案
按热度按时间pu82cl6c1#
只是做了一些小的改动使它工作起来