kafka配置命令

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>cn.itzkx.Kafka_all</groupId> <artifactId>Kafka_all</artifactId> <packaging>pom</packaging> <version>1.0-SNAPSHOT</version> <modules> <module>kafka</module> </modules>
</project>

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <parent> <artifactId>Kafka_all</artifactId> <groupId>cn.itzkx.Kafka_all</groupId> <version>1.0-SNAPSHOT</version> </parent> <modelVersion>4.0.0</modelVersion> <artifactId>kafka</artifactId>

<dependencies> <dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-clients</artifactId> <version>0.11.0.1</version> </dependency>
</dependencies>
</project>

package cn.itzkx.kafka.consumer;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.Arrays;
import java.util.Properties;

/**
 * @ClassName OrderConsumer
 * @Description TODO 订单的消费者，获取数据，获取到数据即可以将数据保存到订单表中
 */
public class OrderConsumer { public static void main(String[] args) { //创建连接对象 Properties props = new Properties(); //消费者配置参数 props.put("bootstrap.servers", "node01:9092");//连接到哪个kafka集群 props.put("group.id", "test1");//设置消费者组 props.put("enable.auto.commit", "true");//自动提交，offset是否会自动保存到kafka的对应的topic上 props.put("auto.commit.interval.ms", "1000");//每1秒钟提交一次偏移量 props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(props); //消费数据 //订阅topic 参数是多个topic的一个集合 kafkaConsumer.subscribe(Arrays.asList("spider-test")); while (true){ //拉去100毫秒内的数据 ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(100);//拉去数据 for (ConsumerRecord<String, String> consumerRecord : consumerRecords) { //打印数据 System.out.println("消费到的数据："+consumerRecord.value()); } } }
}

package cn.itzkx.kafka.consumer;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.Arrays;
import java.util.Properties;

/**
 * @ClassName OrderConsumer
 * @Description TODO 订单的消费者，获取数据，获取到数据即可以将数据保存到订单表中
 */
public class OrderConsumer { public static void main(String[] args) { //创建连接对象 Properties props = new Properties(); //消费者配置参数 props.put("bootstrap.servers", "node01:9092");//连接到哪个kafka集群 props.put("group.id", "test1");//设置消费者组 props.put("enable.auto.commit", "true");//自动提交，offset是否会自动保存到kafka的对应的topic上 props.put("auto.commit.interval.ms", "1000");//每1秒钟提交一次偏移量 props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(props); //消费数据 //订阅topic 参数是多个topic的一个集合 kafkaConsumer.subscribe(Arrays.asList("spider-test")); while (true){ //拉去100毫秒内的数据 ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(100);//拉去数据 for (ConsumerRecord<String, String> consumerRecord : consumerRecords) { //打印数据 System.out.println("消费到的数据："+consumerRecord.value()); } } }
}