Apache Kafka知识点表格总结

      之前的项目中用到RabbitMQ比较多，也有用到RocketMQ,，虽然项目中没有用到过Kafka，不过自己在空闲时间学习过,而且在面试中也会问到，因为还是有不少公司用到Kafka，所以做个总结，一方面是做为面试参考，还有就是以后项目如果用到，也可以作为实践参考。

     Apache Kafka是一个分布式流处理平台，主要用于构建实时数据管道和流式应用程序。它具有高吞吐量、可扩展性、持久性、容错性等特点，并且能够高效地处理高负载数据。

目录

1.Kafka 基本概念

2. Kafka 架构

3.Kafka 数据持久化

4. Kafka 消息传递保证

 5.Kafka 消费者API

6. Kafka 生产者API

7. Kafka 连接器和流处理 (Kafka Connect & Kafka Streams)

8，Kafka安全性

 9.Kafka 监控和运维 (Kafka Monitoring & Operations)

10. Kafka 集群管理 (Kafka Cluster Management)

11. Kafka 性能优化 (Kafka Performance Optimization)

12. Kafka 集成和生态系统 (Kafka Integration & Ecosystem)

13. Kafka 高级特性 (Kafka Advanced Features)

14. Kafka 消息流处理模式 (Kafka Stream Processing Patterns)

15.Kafka 消息序列化与反序列化

16. Kafka 消息传递语义 (Kafka Messaging Semantics)

17. Kafka 集群扩展性 (Kafka Cluster Scalability)

18. Kafka 测试和调试 (Kafka Testing & Debugging)

19.Spring Boot整合Kafka

    1.Kafka 基本概念

2. Kafka 架构

3.Kafka 数据持久化

4. Kafka 消息传递保证

 5.Kafka 消费者API

6. Kafka 生产者API

7. Kafka 连接器和流处理 (Kafka Connect & Kafka Streams)

8，Kafka安全性

 9.Kafka 监控和运维 (Kafka Monitoring & Operations)

10. Kafka 集群管理 (Kafka Cluster Management)

11. Kafka 性能优化 (Kafka Performance Optimization)

12. Kafka 集成和生态系统 (Kafka Integration & Ecosystem)

13. Kafka 高级特性 (Kafka Advanced Features)

14. Kafka 消息流处理模式 (Kafka Stream Processing Patterns)

15.Kafka 消息序列化与反序列化

16. Kafka 消息传递语义 (Kafka Messaging Semantics)

17. Kafka 集群扩展性 (Kafka Cluster Scalability)

18. Kafka 测试和调试 (Kafka Testing & Debugging)

19.Spring Boot整合Kafka

Spring Boot整合Kafka主要涉及以下几个步骤：

1. 添加依赖：在Spring Boot项目的pom.xml文件中添加Spring Kafka的相关依赖。

2. 配置Kafka：在application.properties或application.yml中配置Kafka的连接信息，包括Bootstrap Servers。

3. 创建Producer配置：配置Kafka Producer并创建Kafka Template，用于发送消息。

4. 创建Consumer配置：配置Kafka Consumer，创建用于接收消息的监听器。

5. 发送和接收消息：使用Kafka Template发送消息，使用@KafkaListener注解或消息监听器容器来接收消息。

6. 管理监听器：可以控制监听器的启动、停止和恢复。

7. 高级特性：使用Spring Kafka提供的高级特性，如事务性消息、自定义分区器、消息转发等

具体步骤：

       首先需要安装Kafka服务,网上教程挺多，确保有对应的服务。

      19.1 添加依赖：

    在pom.xml中添加如下依赖：

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
    <version>2.2.6.RELEASE</version>
</dependency>

    19.2 配置Kafka

   在application.properties中添加：

spring:
  kafka:
    consumer:
      key-deserializer:  org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer:  org.apache.kafka.common.serialization.StringDeserializer
      group-id: myGroup
      enable-auto-commit: false
      auto-commit-interval: 1000
      max-poll-records: 1
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      properties:
        acks: all
        retries: 3
        batch.size: 16384
        linger.ms: 1
        buffer.memory: 33554432
        max.block.ms: 6000
        compression.type: none
        max.request.size: 1048576
    bootstrap-servers: 127.0.0.1:9092

   19.3  创建生产者：发送消息到Kafka

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
 
@Service
public class KafkaProducer {
 
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
 
    public void sendMessage(String topic, String message) {
        kafkaTemplate.send(topic, message);
    }
}

 19.4  创建消费者：从Kafka消费消息

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
 
@Component
public class KafkaConsumer {
 
    @KafkaListener(topics = "myTopic", groupId = "myGroup")
    public void listen(String message) {
        System.out.println("Received message in group myGroup: " + message);
    }
}

19.5 启动类：确保@EnableKafka注解已经添加

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.kafka.annotation.EnableKafka;
 
@SpringBootApplication
@EnableKafka
public class KafkaApplication {
 
    public static void main(String[] args) {
        SpringApplication.run(KafkaApplication.class, args);
    }
}