Windows下快速启动Kafka以及三种发送消息的方式_kafka windows

目录

一、下载Kafka

二、启动kafka

2.1 启动kafka前得先启动zk

2.2 启动kafka

三、操作Kafka

3.1 创建 Kafka 主题（Topic）

3.2 将信息写入主题（Topic）

3.3 读取信息

四、Java实践-三种发送消息的方式

4.1 异步发送-无回调

4.2 异步发送-有回调

4.3 同步发送

---

Kafka是一种分布式的基于发布/订阅的消息系统，它的高吞吐量、灵活的offset是其它消息系统所没有的。

kafka优缺点

优点:

• ​ 基于磁盘的数据存储 ​

• 高伸缩性 ​

• 高性能

• 应用场景 : 收集指标和日志 ​ 提交日志 流处理

缺点:

• ​ 运维难度大 ​
• 偶尔有数据混乱的情况 ​
• 对zookeeper强依赖 ​
• 多副本模式下对带宽有一定要求

一、下载Kafka

访问 Apache Kafka 下载最新的 Kafka 版本并解压

解压后如下：

 kafka依赖于zookeeper，但是我们不必去下载zk,已经内置了一个

二、启动kafka

2.1 启动kafka前得先启动zk

E:\study\kafka_2.12-3.4.0\bin\windows>zookeeper-server-start.bat ../../config/zookeeper.properties

2.2 启动kafka

E:\study\kafka_2.12-3.4.0\bin\windows>kafka-server-start.bat ../../config/server.properties

三、操作Kafka

3.1 创建 Kafka 主题（Topic）

我们可执行如下命令用于查看已经存在的主题：

E:\study\kafka_2.12-3.4.0\bin\windows>kafka-topics.bat --list --bootstrap-server localhost:9092

用于刚起，所以并不存在任何主题 。

我们来创建一个名为 quick_start_topic 的主题：

 
E:\study\kafka_2.12-3.4.0\bin\windows>kafka-topics.bat --create --topic quick_start_topic --bootstrap-server localhost:9092
WARNING: Due to limitations in metric names, topics with a period ('.') or underscore ('_') could collide. To avoid issues it is best to use either, but not both.
Created topic quick_start_topic.

再来查看存在的主题：

E:\study\kafka_2.12-3.4.0\bin\windows>kafka-topics.bat --list --bootstrap-server localhost:9092
quick_start_topic

查看详细信息：

E:\study\kafka_2.12-3.4.0\bin\windows>kafka-topics.bat --describe --topic quick_start_topic --bootstrap-server localhost:9092
Topic: quick_start_topic        TopicId: TYIVCN3zSaWSvKDhhPvZRQ PartitionCount: 1       ReplicationFactor: 1    Configs:
        Topic: quick_start_topic        Partition: 0    Leader: 0       Replicas: 0     Isr: 0

3.2 将信息写入主题（Topic）

Kafka 客户端通过网络与 Kafka 代理通信以写入（或读取）事件。一旦收到，代理将以持久和容错的方式存储事件，只要您需要——甚至永远。

运行控制台生产者客户端以将一些事件写入到指定主题。默认情况下，输入的每一行都会产生一个单独的事件写入到主题

E:\study\kafka_2.12-3.4.0\bin\windows>kafka-console-producer.bat --topic quick_start_topic --bootstrap-server localhost:9092
>this is the first message
>this is the second message
>终止批处理操作吗(Y/N)? y

ctrl +c 按照提示实在输入y退出命令交互模式

此时我们已经成功的将两条消息发布到了kafla的 quick_start_topic主题中

3.3 读取信息

打开另一个终端会话并运行消费者客户端以读取刚刚创建的事件：

E:\study\kafka_2.12-3.4.0\bin\windows>kafka-console-consumer.bat --topic quick_start_topic --from-beginning --bootstrap-server localhost:9092
this is the first message
this is the second message

命令行的简单演示到此结束，我们来试下java demo

四、Java实践-三种发送消息的方式

正式进入生产者代码实践之前，首先列举出生产者方大致的参数列表如下：

使用boot搭建项目并勾选kafka或者手动引入相关依赖

引入kafka依赖

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

发送消息前，我们使用命令开启消费者监听

E:\study\kafka_2.12-3.4.0\bin\windows>kafka-console-consumer.bat --topic quick_start_topic --from-beginning --bootstrap-server localhost:9092

4.1 异步发送-无回调

发送并忘记的方式本质上也是一种异步的方式，只是它不会获取消息发送的返回结果，这种方式的吞吐量是最高的，但是无法保证消息的可靠性：

package com.cjian.kafkademo.demo;
 
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
 
import java.util.Properties;
 
/**
 * @Author: cjian
 * @Date: 2023/5/16 20:01
 * @Des: 异步发送-无回调
 */
public class AsyncProducerNoCallback {
    public static void main(String[] args) {
        // 1. 创建kafka生产者配置对象
        Properties properties = new Properties();
        // 2. 给 kafka 配置对象添加配置信息：bootstrap.servers
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        // key,value 序列化（必须）：key.serializer，value.serializer
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        // 3. 创建 kafka 生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);
        // 4. 调用 send 方法,异步发送消息，无回调接口
        for (int i = 0; i < 5; i++) {
            kafkaProducer.send(new ProducerRecord<>("quick_start_topic", "producerAsyncMessage-" + i));
        }
        // 5. 关闭资源
        kafkaProducer.close();
    }
 
}

运行后，观察“消费者”窗口： 

E:\study\kafka_2.12-3.4.0\bin\windows>kafka-console-consumer.bat --topic quick_start_topic --from-beginning --bootstrap-server localhost:9092
producerAsyncMessage-0
producerAsyncMessage-1
producerAsyncMessage-2
producerAsyncMessage-3
producerAsyncMessage-4

4.2 异步发送-有回调

异步发送+回调函数(消息以异步的方式发送，通过回调函数返回消息发送成功/失败)

在调用send方法发送消息的同时，指定一个回调函数，服务器在返回响应时会调用该回调函数，通过回调函数能够对异常情况进行处理，当调用了回调函数时，只有回调函数执行完毕生产者才会结束，否则一直会阻塞

package com.cjian.kafkademo.demo;
 
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
 
import java.util.Properties;
 
/**
 * @Author: cjian
 * @Date: 2023/5/16 20:09
 * @Des: 异步发送-有回调
 */
public class AsyncProducerWithCallback {
    public static void main(String[] args) throws InterruptedException {
        // 1. 创建Kafka生产者的配置对象
        Properties properties = new Properties();
        // 2. 给kafka配置对象添加配置信息
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        // 3. key 序列化 key.serializer，value.serializer
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        // 4. value 序列化 value.serializer
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        // 5. 创建kafka生产者对象
        KafkaProducer<String,String> kafkaProducer = new KafkaProducer<String, String>(properties);
        for (int i = 0; i < 5; i++) {
            kafkaProducer.send(new ProducerRecord<>("quick_start_topic", "asyncCallbackMessage-" + i), new Callback() {
                // 该方法在Producer 收到 ack 时调用，为异步调用
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    if (e == null) {
                        // 没有异常，输出信息到控制台
                        System.out.println("topic：" + recordMetadata.topic() + " -> " + " 分区 " + recordMetadata.partition());
                    }else {
                        e.printStackTrace();
                    }
                }
            });
            // 延迟发送是否会发送到不同分区
            Thread.sleep(2);
        }
        // 5. 关闭资源
        kafkaProducer.close();
    }
}

E:\study\kafka_2.12-3.4.0\bin\windows>kafka-console-consumer.bat --topic quick_start_topic --from-beginning --bootstrap-server localhost:9092
this is the first message
this is the second message
producerAsyncMessage-0
producerAsyncMessage-1
producerAsyncMessage-2
producerAsyncMessage-3
producerAsyncMessage-4
asyncCallbackMessage-0
asyncCallbackMessage-1
asyncCallbackMessage-2
asyncCallbackMessage-3
asyncCallbackMessage-4

4.3 同步发送

以同步的方式发送消息时，一条一条的发送，对每条消息返回的结果判断， 可以明确地知道每条消息的发送情况，但是由于同步的方式会阻塞，只有当消息通过get返回future对象时，才会继续下一条消息的发送

package com.cjian.kafkademo.demo;
 
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 java.util.Properties;
import java.util.concurrent.ExecutionException;
 
/**
 * @Author: cjian
 * @Date: 2023/5/16 20:12
 * @Des: 同步发送
 */
public class SyncProducer {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 1. 创建kafka生产者配置对象
        Properties properties = new Properties();
        // 2. 给 kafka 配置对象添加配置信息：bootstrap.servers
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        // key,value 序列化（必须）：key.serializer，value.serializer
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        // 3. 创建 kafka 生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);
        // 4. 调用 send 方法,发送消息
        for (int i = 0; i < 5; i++) {
            RecordMetadata recordMetadata = kafkaProducer.send(new ProducerRecord<>("quick_start_topic", "syncMessage-" + i)).get();
            System.out.println("send response result,topic：" + recordMetadata.topic() + ", " + " 分区:" + recordMetadata.partition());
        }
        // 5. 关闭资源
        kafkaProducer.close();
    }
}

E:\study\kafka_2.12-3.4.0\bin\windows>kafka-console-consumer.bat --topic quick_start_topic --from-beginning --bootstrap-server localhost:9092
this is the first message
this is the second message
producerAsyncMessage-0
producerAsyncMessage-1
producerAsyncMessage-2
producerAsyncMessage-3
producerAsyncMessage-4
asyncCallbackMessage-0
asyncCallbackMessage-1
asyncCallbackMessage-2
asyncCallbackMessage-3
asyncCallbackMessage-4
syncMessage-0
syncMessage-1
syncMessage-2
syncMessage-3
syncMessage-4