我已经使用kafkatemplate bean将sender类设置为将有效负载发送给主题在SenderConfiguration类中进行了一些配置。
Sender类别
@Component
public class Sender {
private static final Logger LOGGER = (Logger) LoggerFactory.getLogger(Sender.class);
@Autowired
private KafkaTemplate<String, String> kafkaTemplate;
public void send(String topic, String payload) {
LOGGER.info("sending payload='{}' to topic='{}'", payload, topic);
kafkaTemplate.send(topic, "1", payload);
}
}
,senderConfiguration类
@Configuration
public class SenderConfig {
@Value("${kafka.bootstrap-servers}")
private String bootstrapServers;
@Bean
public Map<String, Object> producerConfigs() {
Map<String, Object> props = new HashMap<>();
// list of host:port pairs used for establishing the initial connections to the Kakfa cluster
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
return props;
}
@Bean
public ProducerFactory<String, String> producerFactory() {
return new DefaultKafkaProducerFactory<>(producerConfigs());
}
@Bean
public KafkaTemplate<String, String> kafkaTemplate() {
return new KafkaTemplate<>(producerFactory());
}
@Bean
public Sender sender() {
return new Sender();
}
}
问题出在非生产中
此处为application.yml文件属性
kafka:
bootstrap-servers: localhost:9092
topic:
helloworld: helloworld.t
并且只是包含]的控制器>
@RestController public class Controller { protected final static String HELLOWORLD_TOPIC = "helloworld.t"; @Autowired private Sender sender; @RequestMapping("/send") public String SendMessage() { sender.send(HELLOWORLD_TOPIC, "message"); return "success"; } }
且例外是
2017-12-20 09:58:04.645 INFO 10816 --- [nio-7060-exec-1] o.a.kafka.common.utils.AppInfoParser : Kafka version : 0.10.1.1
2017-12-20 09:58:04.645 INFO 10816 --- [nio-7060-exec-1] o.a.kafka.common.utils.AppInfoParser : Kafka commitId : f10ef2720b03b247
2017-12-20 09:59:04.654 ERROR 10816 --- [nio-7060-exec-1] o.s.k.support.LoggingProducerListener : Exception thrown when sending a message with key='1' and payload='message' to topic helloworld.t:
org.apache.kafka.common.errors.TimeoutException: Failed to update metadata after 60000 ms.
我已经使用kafkatemplate bean使sender类通过SenderConfiguration类中的某些配置将有效负载发送到主题。发送方类@Component公共类发送方{private static ...
使用包含密钥的方法
这意味着您的经纪人没有运行。检查server.log并在必要时重新启动代理
这种错误的可能性很小。