kafka笔记（1）-CFANZ编程社区

文章目录

目前md文件尚未上传到git仓库中，如果需要md文件的可以私信我

kafka中的结构
在这里插入图片描述

组件	是什么
Producer	消息生产者：向Broker发送消息的客户端
Producer	消息消费者：从Broker读取消息的客户端
Broker	消息中间间处理节点：一个Kafka节点就是一个Broker，一个或多个Broker可以组成一个Kafka集群（一般集群建议奇数台，最少三台）
Topic	主题：发布到Kafka集群的每条消息都需要指定一个Topic，Kafka的消息都是保存在Topic中
Partition	分区：物理上的存储概念，一个Topic可以分为多个Partition，每个Partition内部都是有序的，但是多个分区组合起来的数据并不是有序的
Consumer Group	消费者组：每个Consumer属于一个特定的Consumer Group（CG），一条消息可以发送到多个不同的CG，但同一个分区的数据只有该CG中的一个Consumer能消费该数据

1，消息队列的两种模式

这个和rabbitmq是不一样的，mq的模式要多于kafka的，卡夫卡分为两种模式，一种是点对点，一种是发布订阅的模式

点对点模式

当收到消费者的确定的时候，是会删除消息的
在这里插入图片描述

发布/订阅模式

在这里插入图片描述

2，kafka的基础架构

生产者----->>kafka----->>消费者

1,kafka可以创建集群，创建多个broker（分散数据的大小）

2，消费者也可以创建组的形式，一个组多个消费者，提高消费的速度。**注意：**同个组内的消费者不能消费同一个partition

3，为了可用，每个partition增加若干副本

在这里插入图片描述

3，kafka的命令行操作

在这里插入图片描述

4，kafka的发送过程

（1）首先有个main线程，启动生产者，调用send（）进行发送，这个时候，可以配置拦截器，一般是使用flume中的拦截器

（2）调用序列化器，方便在网上传输，不是使用的Java中的序列化，应为java中的序列化太过于繁重，需要大量的内容来保证验证等

（3）采用分区器，吧数据分布到不同的队列上，这是基于内存上的，默认的总的大小是32m。每一个小块默认是16kb

（4）当达到batch.size，或者linger.size中的任意一个的时候，就会有一个sender线程去读取，进行转发。

5，Idea编写kafka

（1），导入依赖

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>org.example</groupId>
    <artifactId>kafka</artifactId>
    <version>1.0-SNAPSHOT</version>
    <properties>
        <maven.compiler.target>1.8</maven.compiler.target>
        <maven.compiler.source>1.8</maven.compiler.source>
        <junit.version>5.7.0</junit.version>
    </properties>
<dependencies>
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
        <version>3.0.0</version>
    </dependency>


    </dependencies>
</project>

(2),编写代码

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.common.serialization.StringSerializer;

import java.util.Properties;

public class test01 {
    public static void main(String[] args) {
        //配置
        Properties properties=new Properties();
        //连接集群
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"主机名:9092");
        //指定的keycalue进行序列化
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
        //创建kafka对象
        KafkaProducer<String,String> kafkaProducer=new KafkaProducer<String, String>(properties);
        //发送数据
        for(int i=0;i<5;i++){
            //linux中要建立first这个topic，而且要开启kafka
            kafkaProducer.send(new ProducerRecord<>("first","dadaobajiao"+i));
            //kafkaProducer.send(new ProducerRecord<>("first","dadaobajiao"+i)，new callback{重写});，这是含有回调的异步
            // kafkaProducer.send(new ProducerRecord<>("first","dadaobajiao"+i)).get();
            //这是同步发送
        }
        //关闭资源
        kafkaProducer.close();
    }
}

6，kafka分区器的优点

$[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-h9aLl7Fz-1651303376078)(C:\Users\打倒芭蕉\AppData\Roaming\Typora\typora-user-images\image-20220429105825192.png)]$

7，分区的策略

主要有三种分区策略

（1）指明 partition 的情况下，直接将指明的值直接作为 partiton 值；

（2）没有指明 partition 值但有 key 的情况下，将key 的 hash 值与topic 的 partition数进行取余得到partition 值；

（3）既没有 partition 值又没有 key 值的情况下，第一次调用时随机生成一个整数（后面每次调用在这个整数上自增），将这个值与 topic 可用的 partition 总数取余得到 partition值，也就是常说的round-robin轮询算法。

在这里插入图片描述

8，kafka的可靠传输

对于可靠性是由三种，分别是0，1，-1

其中-1可靠性最强，但是效率是最低的。（会发生重复传输）

使完全可靠可以建立一个ISR队列

在这里插入图片描述

a，解决重复传输----幂等性

• 至少一次（At Least Once）= ACK级别设置为-1 + 分区副本大于等于2 + ISR里应答的最小副本数量大于等于2

• 最多一次（At Most Once）****= ACK级别设置为0

• 总结：

At Least Once可以保证数据不丢失，但是不能保证数据不重复；

At Most Once可以保证数据不重复，但是不能保证数据不丢失。

• **精确一次（Exactly Once）：**对于一些非常重要的信息，比如和钱相关的数据，要求数据既不能重复也不丢失。

Kafka 0.11版本以后，引入了一项重大特性：幂等性和事务。

b，幂等性的原理

幂等性就是指Producer不论向Broker发送多少次重复数据，Broker端都只会持久化一条，保证了不重复。

精确一次（Exactly Once） = 幂等性 + 至少一次（ ack=-1 + 分区副本数>=2 + ISR最小副本数量**>=2**）。

注意：幂等性在多个会话中，不能保证消息不会重复

重复数据的判断标准：具有**<PID, Partition, SeqNumber>**相同主键的消息提交时，Broker只会持久化一条。其

中PID是Kafka每次重启都会分配一个新的；Partition 表示分区号；Sequence Number是单调自增的。

所以幂等性只能保证的是在单分区单会话内不重复。

2）如何使用幂等性

开启参数 enable.idempotence 默认为 true，false 关闭。

c，事务

开启事务，必须要开启幂等性

$[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-6HkuWC4s-1651303376081)(C:\Users\打倒芭蕉\AppData\Roaming\Typora\typora-user-images\image-20220430151600570.png)]$

事务的API

// 1 初始化事务
void initTransactions();
// 2 开启事务
void beginTransaction() throws ProducerFencedException;
// 3 在事务内提交已经消费的偏移量（主要用于消费者）
void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets,
 String consumerGroupId) throws 
ProducerFencedException;
// 4 提交事务
void commitTransaction() throws ProducerFencedException;
// 5 放弃事务（类似于回滚事务的操作）
void abortTransaction() throws ProducerFencedException;

代码，其中注意的使一定要对事务指定id

public static void main(String[] args) {

        // 0 配置
        Properties properties = new Properties();

        // 连接集群 bootstrap.servers
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092,hadoop103:9092");

        // 指定对应的key和value的序列化类型 key.serializer
//        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // 指定事务id
        properties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "tranactional_id_01");

        // 1 创建kafka生产者对象
        // "" hello
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
        //初始化事务
        kafkaProducer.initTransactions();
        //开启事务
        kafkaProducer.beginTransaction();
        
        try {
            // 2 发送数据
            for (int i = 0; i < 5; i++) {
                kafkaProducer.send(new ProducerRecord<>("first", "atguigu" + i));
            }
            //模拟事务出现异常
            int i = 1 / 0;
            //提交事务
            kafkaProducer.commitTransaction();
        } catch (Exception e) {
            kafkaProducer.abortTransaction();
        } finally {
            // 3 关闭资源
            kafkaProducer.close();
        }