kafka 安装快速入门

直接上干货，我们公司最近要进行消息推送指定软件kafka,直接走起。

1.下载

kafka 是apache的项目。下载地址：kafka.apache.org/

点击download kafka 进入查看相关版本进行下载。

我这里用的版本比窘旧一点，公司技术一切求稳。

下载好安装包就已经实现了。

2.安装

说安装就是免安装版本，简单配置一下就可以使用了

直接解压就可以，但是需要我们修改一些配置文件。

kafka 集成了 zookeeper 的软件包，不需要安装，配置就好。

1）：配置zookeeper.properties

进入到config文件夹里面，找到zookeeper.properties文件，进行编辑，找到dataDir，修改这个参数的路径为：

dataDir=D:\kafka\kafka_2.13-3.2.1\zookeeper-data (安装目录\zookeeper-data)

2）：配置server.properties

进入到config文件夹里面，找到server.properties文件，进行编辑，找到log.dirs，修改这个参数的路径为：

log.dirs=D:\kafka\kafka_2.13-3.2.1\kafka-logs (安装目录\kefke-log)

完成了配置，现在就可以启动了

3.启动

1）：启动zookeeper

打开cmd，进入kafka安装目录，执行：

.\bin\windows\zookeeper-server-start.bat .\config\zookeeper.properties

2）：启动kafka

再打开一个cmd，进入kafka安装目录，执行：

.\bin\windows\kafka-server-start.bat .\config\server.properties

完成启动就可以访问了，但是我们访问有端口限制，所以我又进行了端口修改

kafka默认端口号：9092 我这里给它改为8081 方便测试访问。

4.配置端口（可以不修改）

config下

1、service.properties (添加) port = 8081 不指定的话，按照默认9092

2、connect-distributed.properties (集群) bootstrap.servers=localhost:8081

3、producer.properties (生产者) bootstrap.servers=localhost:8081

4、connect-standalone.properties (单机) bootstrap.servers=localhost:8081

5、consumer.properties (消费者) bootstrap.servers=localhost:8081

根据情况配置端口

5.编码

maven坐标

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

1）生产者代码

package com.java;import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.common.serialization.StringSerializer;import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;public class MyKafKaProducer extends Thread{KafkaProducer<Integer, String> producer;String topic; // 主题public MyKafKaProducer(String topic) {// 构建连接配置Properties properties = new Properties();// 若要配多个服务器，用逗号隔开// 注：服务器要开放端口，若云服务器还要在server.properties配置内网IP和外网IPproperties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.32:8081");properties.put(ProducerConfig.CLIENT_ID_CONFIG, "my-producer");properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class.getName());properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());// 构造Client无非是：new 或 工厂模式producer = new KafkaProducer<Integer, String>(properties);this.topic = topic;}public void run() {int num = 0;String msg = "kafka practice msg: " + num;while (num < 20) {try {// 发送消息send()！！! 同步调用// Future.get()会阻塞，等待返回结果......RecordMetadata recordMetadata = producer.send(new ProducerRecord<>(topic, msg)).get();// 等上面get到结果了，才能执行这里System.out.println(recordMetadata.offset() + "->" + recordMetadata.partition() + "->" + recordMetadata.topic());TimeUnit.SECONDS.sleep(2);num++;} catch (InterruptedException e) {e.printStackTrace();} catch (ExecutionException e) {e.printStackTrace();}}}public static void main(String[] args) {// 传入test主题new MyKafKaProducer("test").start();}
}

2）消费者代码

package com.java;import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.IntegerDeserializer;
import org.apache.kafka.common.serialization.StringDeserializer;import java.time.Duration;
import java.util.Collections;
import java.util.Properties;public class MykafkaConsumer extends Thread{KafkaConsumer<Integer, String> consumer;String topic;public MykafkaConsumer(String topic) {// 构建连接配置，这里是ConsumerConfigProperties properties = new Properties();properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.32:8081");properties.put(ConsumerConfig.CLIENT_ID_CONFIG, "my-consumer");// 反序列化，这里是Deserializerproperties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class.getName());properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());// 以下是Producer没有的配置properties.put(ConsumerConfig.GROUP_ID_CONFIG, "my-gid"); // 要加入的groupproperties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000"); // 超时，心跳properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000"); // 自动提交（批量）properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); // 新group消费为位置consumer = new KafkaConsumer<>(properties);this.topic = topic;}public void run() {// 死循环不断消费消息while (true) {// 绑定订阅主题// 注：Collections.singleton返回一个单元素&不可修改Set集合，// 同样的还有singletonList，singletonMapconsumer.subscribe(Collections.singleton(this.topic));// 接收消息 POLL()！！！ConsumerRecords<Integer, String> consumerRecords = consumer.poll(Duration.ofSeconds(1));// 注：一行的lamada表达式可以不用{}consumerRecords.forEach(record -> System.out.println(record.key() + "->" + record.value() + "->" + record.offset()));}}public static void main(String[] args) {// 拉取test主题的消息new MykafkaConsumer("test").start();}
}

topic 主题必须一致才可以实现生产者和消费者的互通