kafka集群操作-JavaAPI操作

1.2、kafka集群操作-JavaAPI操作

1.2.1、添加依赖1.2.2、生产者代码1.2.3、消费者代码

创建maven工程并添加以下依赖jar包的坐标到pom.xml
下面展示一些 内联代码片。

    
        org.apache.kafka
        kafka-clients
         0.10.0.0
    
    
        org.apache.kafka
        kafka-streams
        0.10.0.0
    




    
        
        
            org.apache.maven.plugins
            maven-compiler-plugin
            3.2
            
                1.8
                1.8
                UTF-8
            
        
    

下面展示一些 内联代码片。

public class OrderProducer {
public static void main(String[] args) throws InterruptedException {

Properties props = new Properties(); props.put("bootstrap.servers", "node01:9092"); props.put("acks", "all");
props.put("retries", 0);
props.put("batch.size", 16384);
props.put("linger.ms", 1);
props.put("buffer.memory", 33554432); 
props.put("key.serializer",
"org.apache.kafka.common.serialization.StringSerializer"); 
props.put("value.serializer",
"org.apache.kafka.common.serialization.StringSerializer");
 KafkaProducer kafkaProducer = new KafkaProducer
(props);
for (int i = 0; i < 1000; i++) {
// 发送数据 ,需要一个producerRecord对象,最少参数 String topic, V value kafkaProducer.send(new ProducerRecord("order", "订单信
息！"+i));
Thread.sleep(100);
}
}
}

使用消费者进行消费kafka当中的数据
下面展示一些 内联代码片。

public class OrderConsumer {
public static void main(String[] args) {
// 1连接集群
Properties props = new Properties(); props.put("bootstrap.servers", "hadoop-01:9092"); props.put("group.id", "test");

//以下两行代码 ---消费者自动提交offset值 props.put("enable.auto.commit", "true"); props.put("auto.commit.interval.ms",  "1000");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
KafkaConsumer kafkaConsumer = new KafkaConsumer
(props);
//		 2、发送数据 发送数据需要，订阅下要消费的topic。	order kafkaConsumer.subscribe(Arrays.asList("order")); while (true) {
ConsumerRecords consumerRecords = kafkaConsumer.poll(100);// jdk queue offer插入、poll获取元素。 blockingqueue put插入原生， take获取元素
for (ConsumerRecord record : consumerRecords) { System.out.println("消费的数据为：" + record.value());
}
}
}
}