Java代码kafka通过Kerberos认证消费数据

今天药忘吃喽~ 2024-04-28 13:08 69阅读 0赞

## Java代码kafka通过Kerberos认证消费数据 ##

### kafka消费数据 ###

package com.beijingbank.rcp.hbase;
    
    import java.io.PrintStream;
    import java.time.Duration;
    import java.util.Collections;
    import java.util.Iterator;
    import java.util.Properties;
    import org.apache.kafka.clients.consumer.ConsumerConfig;
    import org.apache.kafka.clients.consumer.ConsumerRecord;
    import org.apache.kafka.clients.consumer.ConsumerRecords;
    import org.apache.kafka.clients.consumer.KafkaConsumer;
    import org.apache.kafka.common.serialization.StringDeserializer;
    
    public class KafkaConsumerExample
    {
      public static void main(String[] args)
      {
        System.setProperty("java.security.krb5.conf", "kerberos/krb5.conf");
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "aaaa:9092,bbbb:9092,ccc:9092");
        
        properties.setProperty("sasl.jaas.config", "com.sun.security.auth.module.Krb5LoginModule required \n        useKeyTab=true \n        storeKey=true  \n        keyTab=\"kerberos/rcp.keytab\" \n        principal=\"hadoop@HADOOP.COM\";");
        
        properties.setProperty("sasl.kerberos.service.name", "kafka");
        properties.setProperty("sasl.mechanism", "GSSAPI");
        properties.setProperty("security.protocol", "SASL_PLAINTEXT");
        properties.setProperty("auto.offset.reset", "earliest");
        properties.setProperty("group.id", "test_003");
        
        Properties properties1 = ConsumerConfig.addDeserializerToConfig(properties, new StringDeserializer(), new StringDeserializer());
        
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer(properties1);
        kafkaConsumer.subscribe(Collections.singleton("rcp_test_002"));
        for (;;)
        {
          ConsumerRecords<String, String> poll = kafkaConsumer.poll(Duration.ofMillis(1000L));
          if (!poll.isEmpty())
          {
            Iterator<ConsumerRecord<String, String>> iterator = poll.iterator();
            while (iterator.hasNext())
            {
              ConsumerRecord<String, String> next = (ConsumerRecord)iterator.next();
              System.out.println((String)next.value());
            }
          }
        }
      }
    }

### kafka发送数据 ###

package com.td.hadoop.hbase;
    
    import java.io.PrintStream;
    import java.util.Properties;
    import java.util.Random;
    import java.util.Scanner;
    import org.apache.kafka.clients.producer.KafkaProducer;
    import org.apache.kafka.clients.producer.ProducerConfig;
    import org.apache.kafka.clients.producer.ProducerRecord;
    import org.apache.kafka.common.serialization.StringSerializer;
    
    public class KafkaProducerExample
    {
      public static void main(String[] args)
      {
        System.setProperty("java.security.krb5.conf", "kerberos/krb5.conf");
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "aaaa:9092,bbbb:9092,cccc:9092");
        
        properties.setProperty("sasl.jaas.config", "com.sun.security.auth.module.Krb5LoginModule required \n        useKeyTab=true \n        storeKey=true  \n        keyTab=\"kerberos/rcp.keytab\" \n        principal=\"hadoop@HADOOP.COM\";");
        
        properties.setProperty("sasl.kerberos.service.name", "kafka");
        properties.setProperty("sasl.mechanism", "GSSAPI");
        properties.setProperty("security.protocol", "SASL_PLAINTEXT");
        
        Properties properties1 = ProducerConfig.addSerializerToConfig(properties, new StringSerializer(), new StringSerializer());
        
        KafkaProducer<String, String> stringStringKafkaProducer = new KafkaProducer(properties1);
        Random random = new Random();
        Scanner scanner = new Scanner(System.in);
        while (scanner.hasNext())
        {
          String next = scanner.next();
          ProducerRecord<String, String> rcp_test_001 = new ProducerRecord("rcp_test_002", random.nextInt() + "", next);
          stringStringKafkaProducer.send(rcp_test_001);
          stringStringKafkaProducer.flush();
          
        }
      }
    }