環境:
SpringBoot 2.1.5.RELEASE
Kafka kafka_2.11-1.1.1.tgz
kafka maven依賴
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
</dependency>
消費者:
public class ConsumerFastStart {
public static final String brokerList = "hadoop000:9092,hadoop001:9092,hadoop002:9092";
public static final String topic = "tests";
public static void main(String[] args) {
Properties properties = new Properties();
properties.put("key.deserializer", StringDeserializer.class.getName());
properties.put("value.deserializer", StringDeserializer.class.getName());
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
properties.put("group.id", args[0]);
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
consumer.subscribe(Collections.singletonList(topic));
while (true){
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
for (ConsumerRecord<String, String> record : records) {
System.out.println(record.value());
}
}
}
}
生産者:
public class ProducerFastStart {
public static final String brokerList = "hadoop000:9092,hadoop001:9092,hadoop002:9092";
public static final String topic = "tests";
public static void main(String[] args) {
Properties properties = new Properties();
properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
//抛出可重試異常,設定重試次數
properties.put(ProducerConfig.RETRIES_CONFIG, 10);
KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
ProducerRecord<String, String> record = new ProducerRecord<>(topic, "Hello, world!");
try {
producer.send(record);
} catch (Exception e) {
e.printStackTrace();
} finally {
producer.close();
}
}
}
問題分析
我在百度這個問題時,很多人都說添加一段日志配置到log4j.properties中,但是一直沒生效。我檢視了maven依賴找到了logback-classic。
我想應該配置的是logback.xml
1.建立logback.xml
2.添加xml配置
<?xml version="1.0" encoding="UTF-8"?>
<configuration>
<logger name="org.apache.kafka.clients" level="info" />
</configuration>
3.重新開機生産者沒有日志輸出,問題解決。
有緣人,希望能幫助你解決這個問題。