kafka实战演练快速上手

package com.hub.producer; import com.sun.javafx.runtime.SystemProperties; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.Producer; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.serialization.StringSerializer; import org.springframework.core.io.support.PropertiesLoaderUtils; import java.io.IOException; import java.io.InputStream; import java.util.Properties; import java.util.concurrent.TimeUnit; public class KafkaProducerDeno { public static void main(String[] args) throws Exception { KafkaProducerDeno kafkaProducerDeno = new KafkaProducerDeno(); kafkaProducerDeno.producer(); } private static Properties getProperties() throws Exception { Properties properties = PropertiesLoaderUtils.loadAllProperties("application.properties"); // Properties properties = new Properties(); // InputStream ips = KafkaProducerDeno.class // .getResourceAsStream(""); // Properties props = new Properties(); // props.load(ips); // // // System.getProperty("bootstrap.servers"); // properties.setProperty("bootstrap.servers", properties.getProperty("bootstrap.servers").toString()); properties.setProperty("group.id",properties.getProperty("group.id").toString()); properties.setProperty("key.deserializer", properties.getProperty("key.deserializer").toString()); properties.setProperty("value.deserializer", properties.getProperty("value.deserializer").toString()); properties.put("key.serializer", StringSerializer.class.getName()); properties.put("value.serializer", StringSerializer.class.getName()); //properties.setProperty("enable.auto.commit", "false"); //earliest latest properties.setProperty("auto.offset.reset", "latest"); properties.setProperty("partition.assignment.strategy", "org.apache.kafka.clients.consumer.RangeAssignor"); //读取topic properties.setProperty("first.uav.topic.name",properties.getProperty("first.uav.topic.name").toString()); //写入topic properties.setProperty("three.area.write.topic",properties.getProperty("three.area.write.topic").toString()); return properties; } /** * @Description: 生产者 */ public static void producer() throws Exception { Properties properties = getProperties(); Producer<String, String> producer = new KafkaProducer<String, String>(properties); int i=0; while(true){ producer.send( new ProducerRecord("test0420", "message: " + ++i)); try { TimeUnit.SECONDS.sleep(1); } catch (InterruptedException e) { e.printStackTrace(); } } } /** * 获取到Kafka的配置详情信息 * * @return * @throws Exception */ private static Properties getKafkaProperties() throws Exception { String postgresqlConfigName = "application.properties"; InputStream stream = Thread.currentThread().getContextClassLoader().getResourceAsStream(postgresqlConfigName); Properties prop = new Properties(); try { prop.load(stream); } catch (IOException e) { throw new Exception("加载配置文件失败，请检查位置文件是否存在===="); } return prop; } }