一、Kafka命令行操作
1.查看当前集群已存在的主题
bin/kafka-topic.sh --zookeeper hd09-01:2181 --list
2.创建主题
bin/kafka-topic.sh --zookeeper hd09-01:2181 --create --relication-factor 3 \
>--partition 1 \
>---topic xinnian
3.删除主题
bin/kafka-topic.sh --zookeeper hd09-01:2181 --delete --topic xinnian
4.启动生产者发送消息(相当于直接创建主题)
bin/kafka-console-producer.sh --broker-list hd09-01:9092 --topic xinnian
5.启动消费者接收消息
bin/kafka-console-consumer.sh --bootstrap-server hd09-01:9092 \
>--topic xinnian
>--from-beginning
6.查看主题的详细信息
bin/kafka-topic.sh --zookeeper hd09-01:2181 --describe --topic xinnian
二、Kafka的常用API(生产者和消费者)
注意！！！在本地编译器上编写kafka的生产者消费者API时，要在本地的hosts文件中添加映射！！
1.生产者
1）配置生产者属性（kafka节点地址，是否等待应答，发送消息失败是否重试，批处理消息大小，批处理数据延迟，内存缓冲，序列化等）生产者发送数据，只需序列化
Properties prop = new Properties();
prop.put("","");
2）实例化生产者
KafkaProducer<String,String> producer = new KafkaProducer<String,String>(prop);
3）生产者发送消息（使用只含有ProducerRecord的send方法）
for(~){
producer.send(new ProducerRecord<String,String>("topic","value"));
}
4）关闭生产者资源
producer.close();
/**
* @author: PrincessHug
* @date: 2019/2/28, 16:36
* @Blog: https://www.cnblogs.com/HelloBigTable/
*/
public class Prodecer {
public static void main(String[] args) {
Properties prop = new Properties(); //参数配置
//kafka节点的地址
prop.put("bootstrap.servers", "192.168.126.128:9092");
//发送消息是否等待应答
prop.put("acks", "all");
//配置发送消息失败重试
prop.put("retries", "");
//配置批量处理消息大小
prop.put("batch.size", "");
//配置批量处理数据延迟
prop.put("linger.ms","");
//配置内存缓冲大小
prop.put("buffer.memory", "");
//消息在发送前必须序列化
prop.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
prop.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); KafkaProducer<String, String> producer = new KafkaProducer<String, String>(prop); for (int i=;i<;i++){
producer.send(new ProducerRecord<String, String>("xinnian", "" + i + "年新年好！"));
} producer.close(); }
}
2.消费者
1）配置消费者属性（服务器地址，消费者组，自动确认偏移量，反序列化）
Properties prop = new Properties();
2）实例化消费者（线程安全，把consumer定义为常量）
final KafkaConsumer<String,String> consumer = new KafkaConsumer<String,String>(prop);
3）释放资源，线程安全
public class Runtime extends Object
每个Java应用程序都有一个Runtime类的Runtime ，允许应用程序与运行应用程序的环境进行接口。
void addShutdownHook(Thread hook) =>注册一个新的虚拟机关机挂钩。
static Runtime getRuntime() =>返回与当前Java应用程序关联的运行时对象。
4）订阅消息主题，消费者拉取消息poll
while(true){
ConsumerRecords<String,String> records = consumer.poll(1000);
//遍历消息
for(ConsumerRecord<String,String> r:records){
System.out.println(r.key() + "----------" + r.value());
}
}
/**
* @author: PrincessHug
* @date: 2019/2/28, 20:07
* @Blog: https://www.cnblogs.com/HelloBigTable/
*/
public class ConsumerDemo {
public static void main(String[] args) {
//配置消费者属性
Properties prop = new Properties();
prop.put("bootstrap.servers","192.168.126.129:9092");
prop.put("group.id","g1");
prop.put("enable.auto.commit","true");
prop.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
prop.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer"); //初始化消费的对象
final KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(prop); //线程安全，释放资源
Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
public void run() {
if (consumer!=null){
consumer.close();
}
}
})); //消费者订阅主题
consumer.subscribe(Arrays.asList("xinnian")); //消费者拉取消息
while (true){
ConsumerRecords<String, String> records = consumer.poll(1000);
for (ConsumerRecord<String, String> record:records){
System.out.println(record.topic()+"-----"+record.value());
}
} }
}