kafka消息系统实战

kafka是什么？

        是一种高吞吐量的、分布式、发布、订阅、消息系统 

1.导入maven坐标

        <dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>2.4.1</version></dependency>

2.编写提供者

public class KafkaProducer {public static void main(String[] args) {Properties prop = new Properties();prop.put("bootstrap.servers", "localhost:9092");prop.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");prop.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");prop.put("acks", "all");prop.put("retries", 0);prop.put("batch.size", 16384);prop.put("linger.ms", 1);prop.put("buffer.memory", 33554432);String topic = "hello"; // 主题org.apache.kafka.clients.producer.KafkaProducer<String, String> producer = new org.apache.kafka.clients.producer.KafkaProducer<>(prop);producer.send(new ProducerRecord<String, String>(topic, Integer.toString(2), "hello kafka"));producer.close();}}

3.编写消费者

public class KafkaConsumer {public static void main(String[] args) throws InterruptedException {Properties prop = new Properties();prop.put("bootstrap.servers", "192.168.8.166:9092");prop.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");prop.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");prop.put("group.id", "con-1");prop.put("auto.offset.reset", "latest");//自动提交偏移量prop.put("auto.commit.intervals.ms", "true");//自动提交时间prop.put("auto.commit.interval.ms", "1000");org.apache.kafka.clients.consumer.KafkaConsumer<String, String> consumer = new org.apache.kafka.clients.consumer.KafkaConsumer<>(prop);ArrayList<String> topics = new ArrayList<>();//可以订阅多个消息topics.add("hello");consumer.subscribe(topics);while (true) {ConsumerRecords<String, String> poll = consumer.poll(Duration.ofSeconds(20));for (ConsumerRecord<String, String> consumerRecord : poll) {System.out.println(consumerRecord);System.out.println(consumerRecord.key());System.out.println(consumerRecord.value());System.out.println(consumerRecord.topic());}}}
}

 4.下载kafka

点此去官网下载——>Apache Kafka

解压后进入config目录 

修改zookeeper.properties

dataDir=D:/kafka_2.13-3.5.1/tmp/zookeeper

修改日志存放的路径server.properties

log.dirs=D:/kafka_2.13-3.5.1/tmp/kafka-logs

 启动zookeeper服务

zookeeper-server-start.bat ../../config/zookeeper.properties

启动kafka服务

kafka-server-start.bat ../../config/server.properties

5.依次启动消费者和生产者，查看发布的消息