一、Kafka安装

1、JDK安装和配置

参考Linux配置Java环境变量 (opens new window)

推荐在Linux上进行Kafka安装和启动

2、下载安装包

官方安装地址 (opens new window)

tar -xzf kafka_2.13-3.4.0.tgz 
cd kafka_2.13-3.4.0/

Kafka控制脚本在Unix和Windows平台有所不同，在Windows平台，请使用 bin\windows\ 而不是bin/, 并将脚本扩展名改为.bat

3、启动服务器

ZooKeeper Service

# linux
$ bin/zookeeper-server-start.sh config/zookeeper.properties

Kafka Service

# linux
$ bin/kafka-server-start.sh config/server.properties

如果需要在后台进行运行，需在上述命令末尾加上&

$ bin/zookeeper-server-start.sh config/zookeeper.properties &
$ bin/kafka-server-start.sh config/server.properties &

4、查看是否启动成功

[root@localhost kafka_2.13-3.4.0]# jps -l
65073 kafka.Kafka
64101 org.apache.zookeeper.server.quorum.QuorumPeerMain
65919 sun.tools.jps.Jps

5、创建topic

创建

$ bin/kafka-topics.sh --create --topic quickstart-events --bootstrap-server localhost:9092

查看描述

$ bin/kafka-topics.sh --describe --topic quickstart-events --bootstrap-server localhost:9092

# output
Topic: quickstart-events        TopicId: j7x6pJqJRaCuxb86QfccJg PartitionCount: 1       ReplicationFactor: 1    Configs: 
        Topic: quickstart-events        Partition: 0    Leader: 0       Replicas: 0     Isr: 0

6、发送消息

$ bin/kafka-console-producer.sh --topic quickstart-events --bootstrap-server localhost:9092
This is my first event
This is my second event

7、接收消息

$ bin/kafka-console-consumer.sh --topic quickstart-events --from-beginning --bootstrap-server localhost:9092
This is my first event
This is my second event

8、docker安装Kafka

拉取镜像

docker pull wurstmeister/zookeeper 
docker pull wurstmeister/kafka

运行服务

zookeeper

docker run -d --restart=always --log-driver json-file --log-opt max-size=100m --log-opt max-file=2  --name zookeeper -p 2181:2181 -v /etc/localtime:/etc/localtime wurstmeister/zookeeper

kafka

docker run -d  --log-driver json-file --log-opt max-size=100m --log-opt max-file=2 --name kafka -p 9092:9092 -e KAFKA_BROKER_ID=0 -e KAFKA_ZOOKEEPER_CONNECT=本机地址:2181/kafka -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://本机地址:9092 -e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092 -v /etc/localtime:/etc/localtime wurstmeister/kafka

查看运行

二、Kafka理论

1、基本作用

目前 Kafka 已经定位为一个分布式流式处理平台，它以高吞吐、可持久化、可水平扩展、支持流数据处理等多种特性而被广泛使用。

Kafka具备三个作用：

• 消息系统：在异步处理、应用解耦、流量削锋和消息通讯四个场景得到应用。
• 存储系统：Kafka 把消息持久化到磁盘，相比于其他基于内存存储的系统而言，有效地降低了数据丢失的风险。
• 流式处理平台：Kafka 不仅为每个流行的流式处理框架提供了可靠的数据来源，还提供了一个完整的流式处理类库，比如窗口、连接、变换和聚合等各类操作。

2、基本概念

一个典型的 Kafka 体系架构包括若干 Producer、若干 Broker、若干 Consumer，以及一个 ZooKeeper 集群。

• Producer：生产者创建信息，为发送消息的一方。
• Consumer：消费者消费信息，为接受信息的一方。
• Broker： 服务代理节点服务代理节点。对于 Kafka 而言，Broker 可以简单地看作一个独立的 Kafka 服务节点或 Kafka 服务实例。

3、主题和分区

在 Kafka 中还有两个特别重要的概念—主题（Topic）与分区（Partition）。Kafka 中的消息以主题为单位进行归类，生产者负责将消息发送到特定的主题（发送到 Kafka 集群中的每一条消息都要指定一个主题），而消费者负责订阅主题并进行消费。

• 主题：Kafka中的消息以主题为单位进行分类的，生产者负责将消息发送到特定的主题，消费者订阅主题来获取消息。

• 分区：同一主题下的不同分区包含的消息是不同的，分区在存储层面可以看作一个可追加的日志（Log）文件，消息在被追加到分区日志文件的时候都会分配一个特定的偏移量（offset）。

  Kafka只能保证分区的有序，不能保证主题的有序。

Kafka 中的分区可以分布在不同的服务器（broker）上，也就是说，一个主题可以横跨多个 broker，以此来提供比单个 broker 更强大的性能。

4、多副本机制

同一分区的不同副本保持的是相同的信息，副本之间是一主多从的关系。

• leader副本负责处理读写请求。
• follower副本只负责于leader副本的消息同步。

副本处于不同的 broker 中，当 leader 副本出现故障时，从 follower 副本中重新选举新的 leader 副本对外提供服务。（类似Redis的哨兵机制）。

三、Kafka开发

1、基础案例

安装依赖

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

生产者

public class Producer {
    public static final String BROKER_LIST = "localhost:9092";
    public static final String TOPIC = "quickstart-events";

    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put("key.serializer",
                "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer",
                "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("bootstrap.servers", BROKER_LIST);


        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
        ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC, "hello, Kafka!");
        try {
            producer.send(record);
        } catch (Exception e) {
            e.printStackTrace();
        }
        producer.close();
    }
}

• 定义一个配置Properties
• 创建一个生产者，并配置
• 生产者往定义的topic发送信息

消费者

public class Consumer {
    public static final String BROKER_LIST = "localhost:9092";
    public static final String TOPIC = "quickstart-events";
    public static final String GROUP_ID = "group.demo";

    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put("key.deserializer",
                "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("value.deserializer",
                "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("bootstrap.servers", BROKER_LIST);
        //设置消费组的名称，具体的释义可以参见第3章
        properties.put("group.id", GROUP_ID);
        //创建一个消费者客户端实例
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
        //订阅主题
        consumer.subscribe(Collections.singleton(TOPIC));
        //循环消费消息
        while (true) {
            ConsumerRecords<String, String> records =
                    consumer.poll(Duration.ofMillis(10000));
            for (ConsumerRecord<String, String> record : records) {
                System.out.println(record.value());
            }
        }
    }
}

• 定义一个配置Properties。
• 创建一个消费者，并配置。
• 消费者订阅topic。
• 消费者获取消息。

2、SpringBoot整合

参考demo-mq-kafka (opens new window)

四、Kafka图形管理

参考kafka-ui-lite (opens new window)

参考

官方文档 (opens new window)

图解Kafka之实战指南 (opens new window)