消息队列Kafka（一）介绍及安装以及基本原理

消息队列Kafka（一）介绍及安装以及基本原理
一、kafka的概述

 1.1 消息中间件的对比

消息中间件对比-选择建议

 1.2 kafka的介绍

kafka是一个分布式流媒体平台，类似于消息队列或企业消息传递系统。kafak官网：
http://kafka.apache.org/
kafka介绍-名词解释：
- producer：发布消息的对象称之为主题的生产者（Kafka topic producer）
- topic：Kafka将消息分门别类，每一个类的消息称之为一个主题（Topic）
- consumer：订阅消息并处理发布的消息的对象称之为主题消费者（consumer）
- broker：已发布的消息保存在一组服务器中，称之为Kafka集群。集群中的每一个服务器都是一个代理（Broker）。消费者可以订阅一个或多个主题，并从broker拉取数据，从而消费这些已发布的消息。
二、Kafka环境的搭建

 2.1 Kafka安装配置

Kafka对于zookeeper是强依赖,保存kafka相关的节点数据，所以安装Kafka之前必须先安装zookeeper
- Docker安装zookeeper
  创建容器
```
docker rm -f zookeeper

docker run -d --name zookeeper --restart=always -p 2181:2181 zookeeper:3.4.14
1
2
3
```
- Docker安装Kafka
  创建容器
```
docker rm -f kafka


docker run -d --name kafka \
--env KAFKA_ADVERTISED_HOST_NAME=192.168.200.130 \
--env KAFKA_ZOOKEEPER_CONNECT=192.168.200.130:2181 \
--env KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://192.168.200.130:9092 \
--env KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092 \
--env KAFKA_HEAP_OPTS="-Xmx256M -Xms256M" \
--restart=always \
--net=host wurstmeister/kafka:2.12-2.3.1
1
2
3
4
5
6
7
8
9
10
11
```
使用kafka-tool连接测试：
点前边的加号
如果是绿色就安装成功

 三、Kafka入门

 3.1生产者
- 生产者发送消息，多个消费者只能有一个消费者接收到消息
- 生产者发送消息，多个消息者都可以接到消息
生产者发送消息
```
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;

/**
 * 生产者
 */
public class ProducerQuickStart100 {

    public static void main(String[] args) {
        //1.kafka的配置信息
        Properties properties = new Properties();
        //kafka的连接地址
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.200.130:9092");
        //发送失败，失败的重试次数
        properties.put(ProducerConfig.RETRIES_CONFIG,5);
        //消息key的序列化器
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
        //消息value的序列化器
properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");

        //2.生产者对象
        KafkaProducer<String,String> producer = new KafkaProducer<String, String>(properties);

        //封装发送的消息
        ProducerRecord<String,String> record = new ProducerRecord<String, String>("itheima-topic","100001","hello kafka");

        //3.发送消息
        producer.send(record);

        //4.关闭消息通道，必须关闭，否则消息发送不成功
        producer.close();
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
```
3.2 消费者

消费者接收消息
```
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

/**
 * 消费者
 */
public class ConsumerQuickStart100 {

    public static void main(String[] args) {
        //1.添加kafka的配置信息
        Properties properties = new Properties();
        //kafka的连接地址
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.200.130:9092");
        //消费者组
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "group2");
        //消息的反序列化器
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

        //2.消费者对象
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);

        //3.订阅主题
        consumer.subscribe(Collections.singletonList("itheima-topic"));

        //当前线程一直处于监听状态
        while (true) {
            //4.获取消息
            ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofMillis(1000));
            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                System.out.println(consumerRecord.key());
                System.out.println(consumerRecord.value());
            }
        }

    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
```
3.3 Kafka的分区机制

测试分区：
进入容器
```
 docker exec -it kafka /bin/bash   
1
```
修改配置文件：
```
 cd /opt/kafka/config/
 vi server.properties 
将num.partitions=1修改为num.partitions=2  
1
2
3
```
修改生产者的代码：
```
  for (int i = 0; i < 5; i++) {
            ProducerRecord<String,String> kvProducerRecord = new ProducerRecord<String,String>("test",String.valueOf(i),"hello kafka");
            producer.send(kvProducerRecord);
        }
1
2
3
4
```
启动两个消费者测试：

传递key的情况下，会按照key进行hash计算

 四、高可用设计方案

 4.1 kafka高可用设计

 4.1.1 集群：
- Kafka的服务器端由被称为Broker的服务进程构成，即一个Kafka集群由多个Broker组成
- 这样如果集群中某一台机器宕机，其他机器上的Broker也依然能够对外提供服务。这其实就是Kafa提供高可用的手段之一
4.1.2 备份机制

Kafka中消息的备份又叫做副本
Kafka定义了两类副本：
- 领导者副本（Leader Replica）
- 追随者服务（Follower Replica）
同步方式：
ISR需要同步复制保存的follower
如果leader失效后，需要选出新的leader，选举的原则如下：
第一：选举时优先从ISR中选定，因为这个列表中follower的数据是与leader同步的
第二：如果ISR列表中的follower都不行了，就只能从其他follower中选举

极端情况，就是所有副本都失效了，这时有两种方案
第一：等待ISR中的一个活过来，选为Leader，数据可靠，但是活过来的时间不确定
第二：选择第一个活过来的Replication，不一定是ISR中的，选为leader，以最快的速度恢复可用性，但数据不一定完整

 4.2 Kafka生产者详情

 4.2.1 消息发送类型
- 同步发送
  使用send（）方法发送，它会返回一个Future对象，调用get（）方法进行等待，就可以知道消息是否发送成功
```
RecordMetadata recordMetadata = producer.send(kvProducerRecord).get();
System.out.println(recordMetadata.offset());
1
2
```
- 异步发送
  调用send（）方法，并指定一个回调函数,服务器在返回响应时调用函数
```
//异步消息发送
producer.send(kvProducerRecord, new Callback() {
    @Override
    public void onCompletion(RecordMetadata recordMetadata, Exception e) {
        if(e != null){
            System.out.println("记录异常信息到日志表中");
        }
        System.out.println(recordMetadata.offset());
    }
});
1
2
3
4
5
6
7
8
9
10
```
4.2.2 参数配置

参数详解：
- ack
  代码配置方式：
```
//ack配置  消息确认机制
prop.put(ProducerConfig.ACKS_CONFIG,"all");
1
2
```
参数的选择说明
- retries
  生产者从服务器收到的错误有可能是临时性错误，在这种情况下，retries参数的值决定了生产者可以重发消息的次数，如果达到这个次数，生产者会放弃重试返回错误，默认情况下生产者会在每次重试之间等待100ms
  代码中设置方式：
```
//重试次数
prop.put(ProducerConfig.RETRIES_CONFIG,10);
1
2
```
- 消息压缩
  默认情况下消息发送时是不会被压缩的
  代码中配置方式：
```
//数据压缩
prop.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,"lz4");
1
2
```
nginx 默认可以压缩，gzip

使用压缩可以降低网络传输开销和存储开销，而这往往是向 Kafka 发送消息的瓶颈所在。

压测 -> 压力测试，准备一台服务器，对kafka集群高强度消息发送，查看CPU和内存等资源

jmeter

4.3 消息的有序性

 4.3.1 Kafka消费者详情

消费者组
- 消费者组指的就是由一个或多个消费者组成的群体
- 一个发布在Topic上的消息被分发给次消费者组中的一个消费者
  - 所有消费者都在一个组中，那么这就变成了queue模型
  - 所有消费者都在不同的组中，那么就完全变成了发布-订阅模型
4.3.2消息有序性

应用场景：
- 即时消息中的单对单和群聊,保证双方发送消息发送顺序与接收方的顺序一致
- 充值转账两个渠道在同一个时间进行余额变更，短信通知必须要由顺序
topic分区中消息只能由消费者组中的唯一一个消费者处理，所以消息肯定是按照先后顺序执行的。但是它也仅仅只保证Topic的一个分区顺序处理，不能保证跨分区的消息先后处理顺序。所以，如果你想要顺序的处理Topic的所有消息，那就只能提供一个分区。

4.4 手动提交偏移量

 4.4.1 提交和偏移量

kafka不会想其他JMS队列那样需要得到消费者的确认，消费者可以使用kafka来追踪消息在分区的位置（偏移量）消费者会在一个叫做_consumer_offset的特殊主题发送消息，消息里包含每个分区的偏移量。如果消费者发生崩溃或有新的消费者加入群组，就会触发再均衡
正常情况：
如果消费者2挂掉以后，会发生在均衡，消费者2负责的分区会被其他消费者消费，在进行均衡，不可避免会出现一些问题。
问题一：
如果提交的偏移量小于客户处理的最后一个消息的偏移量，那么处于两个偏移量之间的消息就会被再次消费。

问题二：
如果提交的偏移量大于客户端的最后一个消息的偏移量，那么处于两个偏移量之间的消息将会丢失。
想要解决问题，还要知道目前kafka提交偏移量的方式：
提交偏移量的方式有两种，分别是自动提交偏移量和手动提交
- 自动提交偏移量
  当enable.auto.commit被设置为true，提交方式就是让消费者自动提交偏移量，每隔五秒消费者会自动把从poll（）方法接收到的最大偏移量提交上去。
- 手动提交，当enable.auto.commit被设置成为false可以有一下三种方式提交
  - 提交当前偏移量（同步提交）
  - 异步提交
  - 同步和异步组合提交
    1.提交当前偏移量
    把enable.auto.commit设置为false,让应用程序决定何时提交偏移量。使用commitSync()提交偏移量，commitSync()将会提交poll返回的最新的偏移量，所以在处理完所有记录后要确保调用了commitSync()方法。否则还是会有消息丢失的风险。
    只要没有发生不可恢复的错误，commitSync（）方法会一直尝试直至提交成功，如果提交失败也可以记录到错误日志中。
```
while (true){
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
    for (ConsumerRecord<String, String> record : records) {
        System.out.println(record.value());
        System.out.println(record.key());
    

    }
        try {
            consumer.commitSync();//同步提交当前最新的偏移量
        }catch (CommitFailedException e){
            System.out.println("记录提交失败的异常："+e);
        }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
```
2.异步提交
手动提交有一个缺点，那就是当发起提交调用时应用会阻塞。当然我们可以减少手动提交的频率，但这个会增加消息重复的概率（和自动提交一样）。另外一个解决办法是，使用异步提交的API。
```
while (true){
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
    for (ConsumerRecord<String, String> record : records) {
        System.out.println(record.value());
        System.out.println(record.key());
    }
    consumer.commitAsync(new OffsetCommitCallback() {
        @Override
        public void onComplete(Map<TopicPartition, OffsetAndMetadata> map, Exception e) {
            if(e!=null){
                System.out.println("记录错误的提交偏移量："+ map+",异常信息"+e);
            }
        }
    });
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
```
3.同步和异步组合提交
异步提交也有个缺点，那就是如果服务器返回提交失败，异步提交不会进行重试。相比较起来，同步提交会进行重试直到成功或者最后抛出异常给应用。异步提交没有实现重试是因为，如果同时存在多个异步提交，进行重试可能会导致位移覆盖。
举个例子，假如我们发起了一个异步提交commitA，此时的提交位移为2000，随后又发起了一个异步提交commitB且位移为3000；commitA提交失败但commitB提交成功，此时commitA进行重试并成功的话，会将实际上将已经提交的位移从3000回滚到2000，导致消息重复消费。
```
try {
    while (true){
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
        for (ConsumerRecord<String, String> record : records) {
            System.out.println(record.value());
            System.out.println(record.key());
        }
        consumer.commitAsync();
    }
}catch (Exception e){+
    e.printStackTrace();
    System.out.println("记录错误信息："+e);
}finally {
    try {
        consumer.commitSync();
    }finally {
        consumer.close();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
```
相关阅读:
敏捷开发的在实际项目中的运用
 网络整体框架介绍
 深度学习31-33
深度学习入门之自动求导（Pytorch）
bat一键给windows server 2012 打补丁
 《Care Bears 爱心熊》人物化身来到 The Sandbox 元宇宙！
二肽Ile-Ala，24787-73-3，H2N-IA-OH
AI实战营第二期第九节《底层视觉与MMEditing》——笔记10
解决node项目一个极度困难的捕获异常却无法读取异常信息的问题
 嵌入式开发：软件启动清单给质量一个良好的开端
原文地址：https://blog.csdn.net/m0_45101736/article/details/126919106

一、kafka的概述

1.1 消息中间件的对比

1.2 kafka的介绍

二、Kafka环境的搭建

2.1 Kafka安装配置

三、Kafka入门

3.1生产者

3.2 消费者

3.3 Kafka的分区机制

四、高可用设计方案

4.1 kafka高可用设计

4.1.1 集群：

4.1.2 备份机制

4.2 Kafka生产者详情

4.2.1 消息发送类型

4.2.2 参数配置

4.3 消息的有序性

4.3.1 Kafka消费者详情

4.3.2消息有序性

4.4 手动提交偏移量

4.4.1 提交和偏移量