kafka ReadFile2Kafka

Posted 2022-01-12 maoxiangyi

public class ReadFile2Kafka 
   public static void main(String[] args) throws Exception 

      if (args.length != 4) 
         throw new RuntimeException("参数列表: bootstrap.servers、topic、sleep、input");
      

      Properties props = new Properties();
      props.put("bootstrap.servers", args[0]);
      props.put("acks", "all");
      props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
      props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
      Producer<String, String> producer = new KafkaProducer<>(props);

      BufferedReader bufferedReader = new BufferedReader(new FileReader(new File(args[3])));
      String line = null;
      while ((line = bufferedReader.readLine()) != null) 
         Thread.sleep(Integer.parseInt(args[2]));
         producer.send(new ProducerRecord<String, String>(args[1], line));
      
      producer.flush();
      producer.close();
      Thread.sleep(3000);
   

shell 脚本

java -jar readFile2Kafka.jar node01:9092 $1  1000 /root/maoxiangyi/streaming/data/$1