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和外网IP
        properties.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) {
        // 构建连接配置，这里是ConsumerConfig
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.32:8081");
        properties.put(ConsumerConfig.CLIENT_ID_CONFIG, "my-consumer");
        // 反序列化，这里是Deserializer
        properties.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"); // 要加入的group
        properties.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，singletonMap
            consumer.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 主题必须一致才可以实现生产者和消费者的互通