1 
      2 
     
       5     
      
       4.0.0
       6  7     
      
       com.ht
       8     
      
       kafkatest
       9     
      
       1.0-SNAPSHOT
      10     
      
       11         
       
        12             
        
         13                 
         
          org.apache.maven.plugins
         14                 
         
          maven-compiler-plugin
         15                 
         
          16                     
          1.717                     
          
           1.7
          18                 
         19             
        20         
       21     
      22 23 24     
      
       25         
       
        26             
        
         org.apache.kafka
        27             
        
         kafka-clients
        28             
        
         0.10.0.0
        29         
       30     
      31 
     

java 代码

1 import org.apache.kafka.clients.CommonClientConfigs; 2 import org.apache.kafka.clients.consumer.ConsumerRecord; 3 import org.apache.kafka.clients.consumer.ConsumerRecords; 4 import org.apache.kafka.clients.consumer.KafkaConsumer; 5  6 import java.util.Collections; 7 import java.util.Properties; 8  9 import static org.apache.kafka.clients.consumer.ConsumerConfig.*;10 11 /**12  * @author sunzq13  * @since 2017/8/2914  */15 public class Application {16     public static void main(String[] args) {17 18         Properties props = new Properties();19         props.put(BOOTSTRAP_SERVERS_CONFIG, "node1:6667,node2:6667,node3:6667,node4:6667");20         props.put(ENABLE_AUTO_COMMIT_CONFIG, "true");21         props.put(GROUP_ID_CONFIG, "test08291103");22 //      props.put(ConsumerConfig.CLIENT_ID_CONFIG, "test0829");23         props.put(AUTO_COMMIT_INTERVAL_MS_CONFIG, 1000);24         props.put(AUTO_OFFSET_RESET_CONFIG, "earliest");25         props.put(KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");26         props.put(VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");27         props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");28 29         KafkaConsumer
     
       consumer = new KafkaConsumer<>(props);30         // topic name: test931         consumer.subscribe(Collections.singleton("test9"));32         while (true) {33             ConsumerRecords
      
        records = consumer.poll(100);34             for (ConsumerRecord
       
         record : records)35                 System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());36         }37     }38 }
       
      
     

启动参数

 -Djava.security.krb5.conf=c:\\app\\conf\\krb5.conf -Djava.security.auth.login.config=c:\\app\\conf\\kafka_jaas.conf  

 

windows 下记得用 \\