kafka Consumer2Local

Posted 2022-01-12 maoxiangyi

public class Consumer2Local 
   private final static Logger log = LoggerFactory.getLogger(Consumer2Local.class);

   public static void main(String[] args) throws IOException 

      if (args.length != 6) 
         throw new RuntimeException("参数列表：bootstrap.servers、topic、groupName,needSecurity、earliest、output " +
               "example: java -jar -Djava.security.auth.login.config=kafka_client_jass.conf kafka_consumer_util.jar bbrd-kafka-133-11:19092 ocenter_xes_order_infos bdc_rt_1 need  earliest /home/hadoop/wangxin/kafka-client/data/ocenter_xes_order_infos"
         );
      

      Properties props = new Properties();
      props.setProperty("bootstrap.servers", args[0]);
      props.setProperty("group.id", args[2]);
      props.setProperty("enable.auto.commit", "true");
      props.setProperty("auto.commit.interval.ms", "1000");
      if ("need".equalsIgnoreCase(args[3])) 
         props.setProperty("security.protocol", "SASL_PLAINTEXT");
         props.setProperty("sasl.mechanism", "PLAIN");
         props.setProperty("sasl.username","xes_data");
         props.setProperty("sasl.password","jrUHKOJJHm6VIQCG");
      
      if ("earliest".equalsIgnoreCase(args[4])) 
         props.setProperty("auto.offset.reset", "earliest");
      
      props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
      props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
      final KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
      consumer.subscribe(Arrays.asList(args[1]));
      final FileWriter fileWriter = new FileWriter(new File(args[5]));
      final BufferedWriter bufferedWriter = new BufferedWriter(fileWriter);
      ConsumerRecords<String, String> records;
      //当虚拟机退出时，暂停消费，然后将数据写入完整。
      Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() 
         @Override
         public void run() 
            try 
               bufferedWriter.flush();
               fileWriter.flush();
               fileWriter.close();
               bufferedWriter.close();
               log.info("安全退出！");
             catch (IOException e) 
               e.printStackTrace();
            
         
      ));

      while (true) 
         records = consumer.poll(100);
         for (ConsumerRecord<String, String> record : records) 
            bufferedWriter.write(record.value());
            bufferedWriter.newLine();
         
      
   

java -jar kafka2local.jar public-common-kafka-1:9092 $1 $1 notNeed  earliest /home/ads/maoxiangyi/kafka_util/data/$1