大数据开发之Kafka（概述、快速入门、生产者）

news2024/11/6 9:36:18

第 1 章：Kafka概述

1.1 定义

Kafka是一个分布式的基于发布/订阅模式的消息队列，主要应用于大数据实时处理领域。
发布/订阅：消息的发布者不会将消息直接发送给特定的订阅者，而是将发布的消息分为不同的类别，订阅者只接收感兴趣的消息。

1.2 消息队列

目前企业中比较常见的消息队列产品主要有Kafka、ActiveMQ、RabbitMQ、RocketMQ等。
在大多数场景主要采用Kafka作为消息队列
在JavaEE开发中主要采用ActiveMQ、RabbitMQ、RocketMQ

1.2.1 传统消息队列的应用场景

1、传统的消费队列的主要应用场景有：缓存/削峰（缓冲）、解耦（少依赖）、异步通信（不必要及时处理）
1）缓存/削峰（缓冲）：有助于控制和优化数据流经过系统的速度，解决生产消息和消费消息的处理速度不一致的情况。
在这里插入图片描述

2）解耦：允许你独立的扩展或修改两边的处理过程，只要确保它们遵循同样的接口约束。
在这里插入图片描述

3）异步通信：允许用户把一个消息放入队列，但并不立即处理它，然后再需要的时候再去处理它们。

在这里插入图片描述

1.2.2 消息队列的两种模式

消息队列主要分为两种模式：点对点模式（一个生产者对口一个消费者）和发布/订阅模式（一对多）
在这里插入图片描述

1.3 Kafka基础框架

在这里插入图片描述

1、Producer：消息生产者，就是向Kafka broker发消息的客户端
2、Consumer：消息消费者，向kafka broker获取消息的客户端
3、Consumer Group（CG）：消费者组，由多个consumer组成。消费者组内每个消费者负责消费不同分区的数据，一个broker可以由多个不同的topic，一个topic下的一个分区只能被一个消费者组内的一个消费者所消费；消费者之间不受影响。消费者组是逻辑上的一个订阅者。
4、Broker：一个kafka服务器就是一个broker。一个broker可以容纳多个不同topic
5、Topic：可以理解为一个队列，生产者和消费者面向的都是一个topic
6、Partition：为了实现扩展性，一个非常大的topic可以分布到多个broker（即服务器）上，一个topic可以分为多个partition，每个partition是一个有序的队列
7、Replica：副本，为保证集群中的某个节点发生故障时，该节点上的partition数据不丢失，且kafka仍然能够继续工作，kafka提供了副本机制，一个topic的每个发你去都有若干个副本，一个leader和若干个follower
8、leader：每个分区副本中的”主“，生产者发送数据的对象，以及消费者消费数据的对象都是leader
9、followeer：每个分区副本中的“从”，实现于leader副本保持同步，在leader发送故障时，称为新的leader

第 2 章：Kafka快速入门

2.1 安装部署

2.1.1 集群部署

2.1.2 集群部署

1、官方下载地址：http://kafka.apache.org/downloads.html
2、上传安装包到102的/opt/software目录下：

[atguigu@hadoop102 software]$ ll
-rw-rw-r--. 1 atguigu atguigu  86486610 3月  10 12:33 kafka_2.12-3.0.0.tgz

3、解压安装包到/opt/module/目录下

[atguigu@hadoop102 software]$ tar -zxvf kafka_2.12-3.0.0.tgz -C /opt/module/

4、进入到/opt/module目录下，修改解压包名为kafka

[atguigu@hadoop102 module]$ mv kafka_2.12-3.0.0 kafka

5、修改config目录下的配置文件server.properties内容如下

[atguigu@hadoop102 kafka]$ cd config/
[atguigu@hadoop102 config]$ vim server.properties
#broker的全局唯一编号，不能重复，只能是数字。
broker.id=102
#处理网络请求的线程数量
num.network.threads=3
#用来处理磁盘IO的线程数量
num.io.threads=8
#发送套接字的缓冲区大小
socket.send.buffer.bytes=102400
#接收套接字的缓冲区大小
socket.receive.buffer.bytes=102400
#请求套接字的缓冲区大小
socket.request.max.bytes=104857600
#kafka运行日志(数据)存放的路径，路径不需要提前创建，kafka自动帮你创建，可以配置多个磁盘路径，路径与路径之间可以用"，"分隔
log.dirs=/opt/module/kafka/datas
#topic在当前broker上的分区个数
num.partitions=1
#用来恢复和清理data下数据的线程数量
num.recovery.threads.per.data.dir=1
# 每个topic创建时的副本数，默认时1个副本
offsets.topic.replication.factor=1
#segment文件保留的最长时间，超时将被删除
log.retention.hours=168
#每个segment文件的大小，默认最大1G
log.segment.bytes=1073741824
# 检查过期数据的时间，默认5分钟检查一次是否数据过期
log.retention.check.interval.ms=300000
#配置连接Zookeeper集群地址（在zk根目录下创建/kafka，方便管理）
zookeeper.connect=hadoop102:2181,hadoop103:2181,hadoop104:2181/kafka

6、配置环境变量

[atguigu@hadoop102 kafka]$ sudo vim /etc/profile.d/my_env.sh
#KAFKA_HOME
export KAFKA_HOME=/opt/module/kafka
export PATH=$PATH:$KAFKA_HOME/bin
[atguigu@hadoop102 kafka]$ source /etc/profile

7、分发环境变量文件并source

[atguigu@hadoop102 kafka]$ xsync /etc/profile.d/my_env.sh
==================== hadoop102 ====================
sending incremental file list

sent 47 bytes  received 12 bytes  39.33 bytes/sec
total size is 371  speedup is 6.29
==================== hadoop103 ====================
sending incremental file list
my_env.sh
rsync: mkstemp "/etc/profile.d/.my_env.sh.Sd7MUA" failed: Permission denied (13)

sent 465 bytes  received 126 bytes  394.00 bytes/sec

total size is 371  speedup is 0.63
rsync error: some files/attrs were not transferred (see previous errors) (code 23) at main.c(1178) [sender=3.1.2]
==================== hadoop104 ====================
sending incremental file list
my_env.sh
rsync: mkstemp "/etc/profile.d/.my_env.sh.vb8jRj" failed: Permission denied (13)

sent 465 bytes  received 126 bytes  1,182.00 bytes/sec
total size is 371  speedup is 0.63
rsync error: some files/attrs were not transferred (see previous errors) (code 23) at main.c(1178) [sender=3.1.2]，
# 这时你觉得适用sudo就可以了，但是真的是这样吗？
[atguigu@hadoop102 kafka]$ sudo xsync /etc/profile.d/my_env.sh
sudo: xsync：找不到命令
# 这时需要将xsync的命令文件，copy到/usr/bin/下，sudo（root）才能找到xsync命令
[atguigu@hadoop102 kafka]$ sudo cp /home/atguigu/bin/xsync /usr/bin/
[atguigu@hadoop102 kafka]$ sudo xsync /etc/profile.d/my_env.sh
# 在每个节点上执行source命令,如何你没有xcall脚本，就手动在三台节点上执行source命令。
[atguigu@hadoop102 kafka]$ xcall source /etc/profile

8、分发安装包

[atguigu@hadoop102 module]$ xsync kafka/

9、修改配置文件的brokerid
分别在hadoop103和104上修改配置文件server.properties中的broker.id=103、broker.id=104
注：broker.id不得重复

[atguigu@hadoop103 kafka]$ vim config/server.properties
broker.id=103	
[atguigu@hadoop104 kafka]$ vim config/server.properties
broker.id=104

10、启动集群
1）先启动Zookeeper集群

[atguigu@hadoop102 kafka]$ zk.sh start

2）一次在102、103、104节点启动kafka

[atguigu@hadoop102 kafka]$ bin/kafka-server-start.sh -daemon config/server.properties	[atguigu@hadoop103 kafka]$ bin/kafka-server-start.sh -daemon config/server.properties	[atguigu@hadoop104 kafka]$ bin/kafka-server-start.sh -daemon config/server.properties

11、关闭集群

[atguigu@hadoop102 kafka]$ bin/kafka-server-stop.sh	
[atguigu@hadoop103 kafka]$ bin/kafka-server-stop.sh	
[atguigu@hadoop104 kafka]$ bin/kafka-server-stop.sh

2.1.4 kafka群起脚本

1、脚本编写
在/home/atguigu/bin目录下创建文件kafka.sh脚本文件：

#! /bin/bash
if (($#==0)); then
  echo -e "请输入参数：\n start  启动kafka集群;\n stop  停止kafka集群;\n" && exit
fi

case $1 in
  "start")
    for host in hadoop103 hadoop102 hadoop104
      do
        echo "---------- $1 $host 的kafka ----------"
        ssh $host "/opt/module/kafka/bin/kafka-server-start.sh -daemon /opt/module/kafka/config/server.properties"
      done
      ;;
  "stop")
    for host in hadoop103 hadoop102 hadoop104
      do
        echo "---------- $1 $host 的kafka ----------"
        ssh $host "/opt/module/kafka/bin/kafka-server-stop.sh"
      done
      ;;
    *)
        echo -e "---------- 请输入正确的参数 ----------\n"
        echo -e "start  启动kafka集群;\n stop  停止kafka集群;\n" && exit
      ;;
esac

2、脚本文件添加权限

[atguigu@hadoop102 bin]$ chmod +x kafka.sh

注意：
停止Kafka集群时，一定要等kafka所有节点进程全部停止后再停止Zookeeper集群。
因为Zookeeper集群当中记录着kafka集群相关信息，Zookeeper集群一旦先停止，Kafka集群就没有办法再获取停止进程的信息，只能手动杀死Kafka进程了。

2.2 Kafka命令行操作

在这里插入图片描述

2.2.1 主题命令行操作

1、查看操作主题命令需要的参数

2、重要的参数如下

参数描述
–bootstrap-server连接kafka Broker主机名称和端口号
–topic操作的topic名称
–create创建主题
–delete删除主题
–alter修改主题
–list查看所有主题
–describe查看主题详细描述
–partitions设置主题分区数
–replication-factor设置主题分区副本
–config更新系统默认的配置

3、查看当前服务器中的所有topic

[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --list

4、创建一个主题名称为first的topic

[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --create --replication-factor 3 --partitions 3 --topic first

5、查看topic的详情

[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic first
Topic: first    TopicId: EVV4qHcSR_q0O8YyD32gFg PartitionCount: 1       ReplicationFactor: 3    Configs: segment.bytes=1073741824
        Topic: first    Partition: 0    Leader: 102     Replicas: 102,103,104   Isr: 102,103,104

6、修改分区数（注意：分区数只能增加，不能减少）

[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --alter --topic first --partitions 3

7、再次查看Topic的详情

[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic first
Topic: first    TopicId: EVV4qHcSR_q0O8YyD32gFg PartitionCount: 3       ReplicationFactor: 3    Configs: segment.bytes=1073741824
        Topic: first    Partition: 0    Leader: 102     Replicas: 102,103,104   Isr: 102,103,104
        Topic: first    Partition: 1    Leader: 103     Replicas: 103,104,102   Isr: 103,104,102
        Topic: first    Partition: 2    Leader: 104     Replicas: 104,102,103   Isr: 104,102,103

8、删除topic

[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --delete --topic first

2.2.2 生产者命令行操作

1、查看命令行生产者的参数

[atguigu@hadoop102 kafka]$ bin/kafka-console-producer.sh

2、重要的参数如下：

参数描述
–bootstrap-server连接kafka Broker主机名称和端口号
–topic操作的topic名称
3、生产消息
[atguigu@hadoop102 kafka]$ bin/kafka-console-producer.sh --broker-list hadoop102:9092 --topic first
>hello world
>atguigu  atguigu

2.2.3 消费者命令行操作

1、查看命令行消费者的参数

[atguigu@hadoop102 kafka]$ bin/kafka-console-consumer.sh

2、重要的参数如下：

参数描述
–bootstrap-server连接kafka Broker主机名称和端口号
–topic操作的topic名称
–from-beginning从头开始消费
–group指定消费者组名称
3、消费消息
[atguigu@hadoop102 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first

4、从头开始消费

[atguigu@hadoop102 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --from-beginning --topic first

思考：再次查看当前kafka中的topic列表，发现了什么？为什么？

第 3 章：Kafka生产者

3.1 生产者消息发送流程

3.1.1 发送原理

Kafka的Producer发送消息采用的是异步发送的方式。
在消息发送的过程中，涉及到了两个线程：main线程和Sender线程，以及一个线程共享变量：RecordAccumulator。
1、main线程中创建了一个双端队列RecordAccumulator，将消息发送给RecordAccumulator。
2、Sender线程不断从RecordAccumulator中拉取消息发送到Kafka broker。
在这里插入图片描述

batch.size：只有数据积累到batch size之后，sender才会发送数据。默认16k
linger.ms：如果数据迟迟未达到batch.size，sender等待linger.ms设置的时间到了之后就会发送数据。单位ms，默认值0ms，表示没有延迟。
0：生产者发送过来的数据，不需要等数据磁盘应答。
1：生产者发送过来的数据，Leader收到数据后应答。
2：-l（all）：生产者发送过来的数据，Leader和SR队列里面的所有节点收起数据后应答。-l和all等价。

3.1.2 生产者重要参数列表

参数名称描述
bootstrap.servers生产者连接集群所需的broker地址清单。可以设置1个或者多个，中间用逗号隔开。生产者从给定的broker里查找到其它broker信息。
key.serializer、value.serializer指定发送消息的key和value的序列化类型。要写全类名。（反射获取）
buffer.memoryRecordAccumulator缓冲区大小，默认32m。
batch.size缓冲区一批数据最大值，默认16k。适当增加该值，可以提高吞吐量，但是如果该值设置太大，会导致数据传输延迟增加。
linger.ms如果数据迟迟未达到batch.size，sender等待linger.time之后就会发送数据。单位ms，默认值是0ms，表示没有延迟。生产环境建议该值大小为5-100ms之间。
acks0：生产者发送过来的数据，不需要等数据落盘应答。1：生产者发送过来的数据，Leader数据落盘后应答。-1（all）：生产者发送过来的数据，Leader和isr队列里面的所有节点数据都落盘后应答。默认值是-1
max.in.flight.requests.per.connection允许最多没有返回ack的次数，默认为5，开启幂等性要保证该值是1-5的数字。
Retries（重试）当消息发送出现错误的时候，系统会重发消息。retries表示重试次数。默认是int最大值，2147483647。如果设置了重试，还想保证消息的有序性，需要设置MAX_IN_flight_requests_per_connection=1否则在重试此失败消息的时候，其它的消息可能发送成功了。
retry.backoff.ms两次重试之间的时间间隔，默认是100ms。
enable.idempotence是否开启幂等性，默认true，开启幂等性。
compression.type生产者发送的所有数据的压缩方式。默认是none，不压缩。支持压缩类型：none、gzip、snappy、lz4和zstd。

3.2 异步发送API

3.2.1 普通异步发送

1、需求：创建Kafka生产者，采用异步的方式发送到Kafka broker
2、异步发送流程如下：
在这里插入图片描述

3、代码编写
1）创建工程kafka-demo
2）导入依赖

<!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients -->
<dependencies>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>3.0.0</version>
        </dependency>
</dependencies>

3）创建包名：com.atguigu.kafka.producer
4）编写代码：不带回调函数的API

package com.atguigu.kafka.producer;

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 CustomProducer {
    public static void main(String[] args) throws InterruptedException {
        // 1. 创建kafka生产者的配置对象
        Properties properties = new Properties();

        // 2. 给kafka配置对象添加配置信息
        properties.put("bootstrap.servers","hadoop102:9092");

        // key,value序列化
        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        // 3. 创建kafka生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);

        // 4. 调用send方法,发送消息
        for (int i = 0; i < 10; i++) {
            kafkaProducer.send(new ProducerRecord<>("first","kafka" + i));
        }

        // 5. 关闭资源
        kafkaProducer.close();
    }
}

5）测试：
在hadoop102上开启kafka消费者

[atguigu@hadoop104 kafka]$ sbin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first

在IDEA中执行上述代码，观察hadoop102消费者输出

[atguigu@hadoop104 kafka]$ sbin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first
kafka0
kafka1
kafka2
kafka3
……

3.2.2 带回调函数的异步发送

1、回调函数callback()会在producer受到ack时调用，为异步屌用。
该方法有两个参数分别是RecordMetadata（元数据信息）和Exception（异常信息）。
1）如果Exception为null，说明消息发送成功。
2）如果Exception不为null，说明消息发送不成功。
2、带回掉函数的异步调用发送流程
在这里插入图片描述

注意：消息发送失败会自动重试，不需要我们在回调函数中手动重试。
3、编写代码：带回调函数的生产者

package com.atguigu.kafka.producer;

import org.apache.kafka.clients.producer.*;
import java.util.Properties;

public class CustomProducerCallback {
    public static void main(String[] args) throws InterruptedException {
        // 1. 创建kafka生产者的配置对象
        Properties properties = new Properties();

        // 2. 给kafka配置对象添加配置信息
        properties.put("bootstrap.servers", "hadoop102:9092");
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092");
        // key,value序列化(必须)
        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        // 3. 创建kafka生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);

        // 4. 调用send方法,发送消息
        for (int i = 0; i < 10; i++) {
            // 添加回调
            kafkaProducer.send(new ProducerRecord<>("first", "kafka" + i), new Callback() {
                // 该方法在Producer收到ack时调用，为异步调用
                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    if (exception == null) 
                        // 没有异常,输出信息到控制台
                        System.out.println("主题"+recordMetadata.topic() +", 分区："+recordMetadata.partition()+", 偏移量："+recordMetadata.offset());
                }
            });
        }

        // 5. 关闭资源
        kafkaProducer.close();
    }
}

4、测试
1）在hadoop102上开启kafka消费者

[atguigu@hadoop104 kafka]$ sbin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first

2）在IDEA中执行代码，观察hadoop102消费者输出

[atguigu@hadoop104 kafka]$ sbin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first
kafka0
kafka1
kafka2
……

3）在IDEA控制台观察回调函数

主题first, 分区：0, 偏移量：10
主题first, 分区：0, 偏移量：11
主题first, 分区：0, 偏移量：12
主题first, 分区：0, 偏移量：13
主题first, 分区：0, 偏移量：14
主题first, 分区：0, 偏移量：15
主题first, 分区：0, 偏移量：16
主题first, 分区：0, 偏移量：17
主题first, 分区：0, 偏移量：18
主题first, 分区：0, 偏移量：19
……

3.3 同步发送API

1、同步发送的意思就是，一条消息发送之后，会阻塞当前线程，直至返回ack。
由于send方法返回的是一个Future对象，根据Future对象的特点，我们也可以实现同步发送的效果，只需要调用Future对象的get方法即可。
2、同步发送流程示意图如下：
在这里插入图片描述

3、编写代码：同步发送消息的生产者

package com.atguigu.kafka.producer;

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;
import java.util.concurrent.ExecutionException;

public class ConsumerProducerSync {
    public static void main(String[] args) throws InterruptedException, ExecutionException {
        // 1. 创建kafka生产者的配置对象
        Properties properties = new Properties();

        // 2. 给kafka配置对象添加配置信息
        //properties.put("bootstrap.servers","hadoop102:9092");
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092");
        // key,value序列化(必须)
        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        // 3. 创建kafka生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);

        // 4. 调用send方法,发送消息
        for (int i = 0; i < 10; i++) {

            // 同步发送
            kafkaProducer.send(new ProducerRecord<>("first","kafka" + i)).get();
        }

        // 5. 关闭资源
        kafkaProducer.close();
    }
}

4、测试
1）在hadoop102上开启kafka消费者

[atguigu@hadoop104 kafka]$ sbin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first

2）在IDEA中执行代码，观察102消费者的消费情况

[atguigu@hadoop104 kafka]$ sbin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first
kafka0
kafka1
kafka2
……

3.4 生产者分区

3.4.1 分区的原因

1、便于合理使用存储资源，每个Partition在一个Broker上存储，可以把海量的数据按照分区切割成一块一块的数据存储在多台Broker上。合理控制分区的任务，可以实现负载均衡的效果。
2、提高并行度，生产者可以以分区为单位发送数据；消费者可以以分区为单位进行消费数据。
在这里插入图片描述

3.4.2 生产者分区策略

1、默认分区器DefaultPartitioner

The default partitioning strategy:
·If a partition is specified in the record, use it
·If no partition is specified but a key is present choose a partition based on a hash of the key
·If no partition or key is present choose the sticky partition that changes when the batch is full.
public class DefaultPartitioner implements Partitioner {
… …
}

2、使用：
1）我们需要将producer发送的数据封装成一个ProducerRecord对象。
2）上述的分区策略，我们在ProducerRecord对象中进行配置。
在这里插入图片描述
在这里插入图片描述

在这里插入图片描述

3）策略实现

代码解释
ProducerRecord(topic,partition_num,…)指明partition的情况下直接发往指定的分区，key的分配方式将无效
ProducerRecord(topic,key,value)没有指明partition值但有key的情况下：将key的hash值与topic的partition个数进行取余得到分区号
ProducerRecord(topic,value)既没有partition值又没有key值得情况下：kafka采用Sticky Partition（黏性分区器），会随机选择一个分区，并尽可能一直使用该分区，待该分区的batch已满或者已完成，kafka再随机一个分区（绝对不会是上一个）进行使用。
3、案例：
1）案例1：将数据发送到指定partition的情况下，如：将所有消息发送到分区1中。
package com.atguigu.kafka.producer;

import org.apache.kafka.clients.producer.*;
import java.util.Properties;

public class CustomProducerCallbackPartitions {
    public static void main(String[] args) {

        // 1. 创建kafka生产者的配置对象
        Properties properties = new Properties();

        // 2. 给kafka配置对象添加配置信息
      properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092");

        // key,value序列化（必须）：
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        // 3. 创建生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
        // 4. 造数据
        for (int i = 0; i < 5; i++) {
            // 指定数据发送到1号分区，key为空（IDEA中ctrl + p查看参数）
            kafkaProducer.send(new ProducerRecord<>("first", 1,"","atguigu " + i), new Callback() {
                @Override
                public void onCompletion(RecordMetadata metadata, Exception e) {
                    if (e == null){
                        System.out.println("主题：" + metadata.topic() + "->"  + "分区：" + metadata.partition()
                        );
                    }else {
                        e.printStackTrace();
                    }
                }
            });
        }

        kafkaProducer.close();
    }
}

2）测试：
（1）在hadoop102上开启kafka消费者

[atguigu@hadoop104 kafka]$ sbin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first

（2）在IDEA中执行代码，观察hadoop102上的消费者消费情况

[atguigu@hadoop104 kafka]$ sbin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first
atguigu0
atguigu1
kafka2
……

（3）观察IDEA中控制台输出

主题：first->分区：1
主题：first->分区：1
主题：first->分区：1
主题：first->分区：1
主题：first->分区：1

3）案例2：没有指明partition但是有key的情况下的消费者分区分配

package com.atguigu.kafka.producer;

import org.apache.kafka.clients.producer.*;
import java.util.Properties;

public class CustomProducerCallbackKey {
    public static void main(String[] args) {
        // 1. 创建配置对象
        Properties properties = new Properties();
        // 2. 配置属性
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092");

        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        // 3. 创建生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
        // 4. 造数据
        for (int i = 1; i < 11; i++) {
            // 创建producerRecord对象
            final ProducerRecord<String, String> producerRecord = new ProducerRecord<>(
                        "first", 
                        i + "",// 依次指定key值为i
                        "atguigu " + i);
            kafkaProducer.send(producerRecord, new Callback() {
                @Override
                public void onCompletion(RecordMetadata metadata, Exception e) {
                    if (e == null){
                        System.out.println("消息："+producerRecord.value()+", 主题：" + metadata.topic() + "->" + "分区：" + metadata.partition()
                        );
                    }else {
                        e.printStackTrace();
                    }
                }
            });
        }

        kafkaProducer.close();
    }
}

4）测试
观察IDEA中控制台输出

消息：atguigu 1, 主题：first->分区：0
消息：atguigu 5, 主题：first->分区：0
消息：atguigu 7, 主题：first->分区：0
消息：atguigu 8, 主题：first->分区：0
消息：atguigu 2, 主题：first->分区：2
消息：atguigu 3, 主题：first->分区：2
消息：atguigu 9, 主题：first->分区：2
消息：atguigu 4, 主题：first->分区：1
消息：atguigu 6, 主题：first->分区：1
消息：atguigu 10, 主题：first->分区：1

3.4.3 自定义分区器

1、生产环境中，我们往往需要更加自由的分区需求，我们可以自定义分区器。
2、需求：在上面的根据key分区案例中，我们发现与我们知道的hash分区结果不同。那么我们就实现一个。
3、实现步骤：
1）定义类，实现Partitioner接口
2）重写partition()方法
4、代码实现

package com.atguigu.kafka.partitioner;

import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import java.util.Map;

/**
 * @author leon
 * @create 2020-12-11 10:43
 * 1. 实现接口Partitioner
 * 2. 实现3个方法:partition,close,configure
 * 3. 编写partition方法,返回分区号
 */
public class MyPartitioner implements Partitioner {
    /**
    *  分区方法
    **/
    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
       // 1. 获取key
        String keyStr = key.toString();
        // 2. 创建分区号，返回的结果
        int partNum;
        // 3. 计算key的hash值
        int keyStrHash = keyStr.hashCode();
        // 4. 获取topic的分区个数
        int partitionNumber = cluster.partitionCountForTopic(topic);
        // 5. 计算分区号
        partNum = Math.abs(keyStrHash) % partitionNumber;
        // 4. 返回分区号
        return partNum;
    }

    // 关闭资源
    @Override
    public void close() {
    }

    // 配置方法
    @Override
    public void configure(Map<String, ?> configs) {
    }
}

5、测试
在生产者代码中，通过配置对象，添加自定义分区器

// 添加自定义分区器
properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG," com.atguigu.kafka.partitioner.MyPartitioner ");

在hadoop102上启动kafka消费者

[atguigu@hadoop102 kafka]$ sbin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first

在IDEA中观察回调信息

消息：atguigu 2, 主题：first->分区：2
消息：atguigu 5, 主题：first->分区：2
消息：atguigu 8, 主题：first->分区：2
消息：atguigu 1, 主题：first->分区：1
消息：atguigu 4, 主题：first->分区：1
消息：atguigu 7, 主题：first->分区：1
消息：atguigu 10, 主题：first->分区：1
消息：atguigu 3, 主题：first->分区：0
消息：atguigu 6, 主题：first->分区：0
消息：atguigu 9, 主题：first->分区：0

3.5 生产经验-生产者如何提高吞吐量

3.5.1 吞吐量

在这里插入图片描述

3.5.2 实例

1、编写代码

package com.atguigu.kafka.producer;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;

public class CustomProducerParameters {

    public static void main(String[] args) throws InterruptedException {

        // 1. 创建kafka生产者的配置对象
        Properties properties = new Properties();

        // 2. 给kafka配置对象添加配置信息：bootstrap.servers
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102: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");

        // batch.size：批次大小，默认16K
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);

        // linger.ms：等待时间，默认0
        properties.put(ProducerConfig.LINGER_MS_CONFIG, 1);

        // RecordAccumulator：缓冲区大小，默认32M：buffer.memory
        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);

        // compression.type：压缩，默认none，可配置值gzip、snappy、lz4和zstd
properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,"snappy");

        // 3. 创建kafka生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);

        // 4. 调用send方法,发送消息
        for (int i = 0; i < 5; i++) {
            kafkaProducer.send(new ProducerRecord<>("first","atguigu" + i));
        }

        // 5. 关闭资源
        kafkaProducer.close();
    }
}

2、测试：
1）在hadoop102上开启kafka消费者

[atguigu@hadoop102 kafka]$ sbin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first

2）在IDEA中执行代码，观察hadoop102上的消费者消费情况

[atguigu@hadoop102 kafka]$ sbin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first
atguigu0
atguigu0
atguigu0
……

3.6 生产经验-数据可靠性

1、回顾消费发送流程
在这里插入图片描述

2、ack应答机制
在这里插入图片描述

3、ack应答级别
在这里插入图片描述

背景：leader收到数据，所有follower都开始同步数据，但有一个follower，因为某种故障，迟迟不能与leader进行同步，那leader就要一直等下去，直到它完成同步，才能发送ack。这个问题怎么解决呢？
Kafka提供的解决方案：ISR队列
1）Leader维护了一个动态的in-sync replica set（ISR）和leader保持同步的follower集合。
2）当ISR中的follower完成数据的同步之后，leader就会给producer发送ack。
3）如果follower长时间（replica.lag.time.max.ms）未向leader同步数据，则该follower将被提出ISR。
Leader发生故障之后，就会从ISR中选举新的leader。
ack应答级别
对于某些不太重要的数据，对数据的可靠性要求不是很高，能够容忍数据的少量丢失，所以没必要等ISR中的follower全部接收成功。
所以Kafka为用户提供了三种可靠性级别，用户根据对可靠性和延迟的要求进行权衡，选择以下的配置。

acks=0这一操作提供了一个最低的延迟，partition的leader副本接收到消息还没有写入磁盘就已经返回ack，当leader故障时有可能丢失数据
acks=1partition的leader副本落盘后返回ack，如果在follower副本同步数据之前leader故障，那么将对丢失数据
acks=-1partition的leader和follower副本全部落盘成功后才返回ack。但是如果在follower副本同步完成后，leader副本所在节点发送ack之前，leader副本发送故障，那么会造成数据重复

4、ack应答机制
在这里插入图片描述

5、案例
代码编写：

package com.atguigu.kafka.producer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;

public class CustomProducerAck {
    public static void main(String[] args) throws InterruptedException {

        // 1. 创建kafka生产者的配置对象
        Properties properties = new Properties();

        // 2. 给kafka配置对象添加配置信息：bootstrap.servers
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092");

        // key,value序列化（必须）：key.serializer，value.serializer
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // 设置acks
        properties.put(ProducerConfig.ACKS_CONFIG, "all");

        // 重试次数retries，默认是int最大值，2147483647
        properties.put(ProducerConfig.RETRIES_CONFIG, 3);

        // 3. 创建kafka生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);

        // 4. 调用send方法,发送消息
        for (int i = 0; i < 5; i++) {
            kafkaProducer.send(new ProducerRecord<>("first","atguigu " + i));
        }

        // 5. 关闭资源
        kafkaProducer.close();
    }
} 

3.7 生产经验-数据去重

3.7.1 数据传递语义

至少一次（At Least Once）=ACK级别设置为-1+分区副本大于等于2+ISR里应答的最小副本数量大于等于2
最多一次（At More Once）=ACK级别设置为0
总结：
1）At Least Once可以保证数据不丢失，但是不能保证数据不重复
2）At More Once可以保证数据不重复，但是不能保证数据不丢失
精确一次（Exactly Once）：对于一些非常重要的信息，比如和钱相关的数据，要求数据既不能重复也不丢失。

3.7.2 幂等性

1、幂等性原理：
幂等性就是指Producer不论向Broker发送多少次重复数据，Broker端都只会持久化一条，保证了不重复。
精确一次（Exactly Once）=幂等性+至少一次（ack=-1+分区副本数>=2+ISR最小副本数量>=2）。
重复数据的判断标准：具有<PID，Partition，SeqNumber>相同主键的消息提交时，Broker只会持久化一条。其中PID是producer每次重启都会分配一个新的：Partition表示分区号；SequenceNumber是单调自增的。
所以幂等性只能保证的是在单分区单会话内不重复。
在这里插入图片描述

2、开启幂等性
在producer的配置对象中，添加参数enable.idempotence，参数值默认为true，设置为false就关闭了。

3.7.3 生产者事务

1、kafka事务原理
在这里插入图片描述

2、事务代码流程

// 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;

3.8 生产经验-数据有序

在这里插入图片描述

3.9生产经验-数据乱序

1、kafka在1.x版本之前保证单分区有序，条件如下：
max.in.flight.requests.per.connection=1（不需要考虑是否开启幂等性）
2、kafka在1.x及以后版本保证数据单分区有序，条件如下：
1）未开启幂等性
max.in.flight.requests.per.connection需要设置为1
2）开启幂等性
max.in.flight.requests.per.connection需要设置小于等于5
原因说明：因为在kafka1.x以后，启用幂等后，kafka服务器会缓存producer发来的最近5个request的元数据，故无论如何，都可以保证最近5个request的数据都是有序的。
在这里插入图片描述

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/1402580.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

性能测试、分析、优化

性能测试、分析、优化

&#x1f345; 视频学习&#xff1a;文末有免费的配套视频可观看 &#x1f345; 关注公众号&#xff1a;互联网杂货铺&#xff0c;回复1 &#xff0c;免费获取软件测试全套资料&#xff0c;资料在手&#xff0c;薪资嘎嘎涨 前言 理论来源于实践又服务于实践&#xff0c;在笔者…
阅读更多...
【算法与数据结构】518、LeetCode零钱兑换 II

【算法与数据结构】518、LeetCode零钱兑换 II

文章目录 一、题目二、解法三、完整代码 所有的LeetCode题解索引&#xff0c;可以看这篇文章——【算法和数据结构】LeetCode题解。 一、题目 二、解法 思路分析&#xff1a;本题的硬币是无数的&#xff0c;因此本题可以抽象成一个完全背包问题。完全背包和01背包的不同之处在于…
阅读更多...
LoadRunner从零开始之接触LoadRunner

LoadRunner从零开始之接触LoadRunner

LoadRunner 是Mercury Interactive 公司开发的一款成熟的性能测试工具&#xff0c;LoadRuner 作为性能测试的实现者&#xff0c;涉及了性能测试流程、性能测试技术和软件 体系架构等众多方面的知识点&#xff0c;可以说&#xff0c;学习LoadRuner 是理解和学习性能测试 的非常好…
阅读更多...
常用设计模式(工厂方法，抽象工厂，责任链，装饰器模式)

常用设计模式(工厂方法，抽象工厂，责任链，装饰器模式)

前言 有关设计模式的其他常用模式请参考 单例模式的实现 常见的设计模式(模板与方法&#xff0c;观察者模式&#xff0c;策略模式) 工程方法 定义 定义一个用于创建对象的接口&#xff0c;让子类决定实例化哪一个类。Factory Method使得一个类的实例化延迟到子类。 ——《设…
阅读更多...
Spring Boot 优雅实现统一数据返回格式+统一异常处理+统一日志处理

Spring Boot 优雅实现统一数据返回格式+统一异常处理+统一日志处理

在我们的项目开发中&#xff0c;我们都会对数据返回格式进行统一的处理&#xff0c;这样可以方便前端人员取数据&#xff0c;当然除了正常流程的数据返回格式需要统一以外&#xff0c;我们也需要对异常的情况进行统一的处理&#xff0c;以及项目必备的日志。 1. 统一返回格式 …
阅读更多...
【小沐学GIS】基于C++绘制三维太阳系SolarSystem（OpenGL、glfw、glut）

【小沐学GIS】基于C++绘制三维太阳系SolarSystem（OpenGL、glfw、glut）

&#x1f37a;三维数字地球系列相关文章如下&#x1f37a;&#xff1a;1【小沐学GIS】基于C绘制三维数字地球Earth&#xff08;OpenGL、glfw、glut&#xff09;第一期2【小沐学GIS】基于C绘制三维数字地球Earth&#xff08;OpenGL、glfw、glut&#xff09;第二期3【小沐学GIS】…
阅读更多...
深度学习基础之数据操作

深度学习基础之数据操作

深度学习中最常用的数据是张量&#xff0c;对张量进行操作是进行深度学习的基础。以下是对张量进行的一些操作&#xff1a; 首先我们需要先导入相关的张量库torch。 元素构造&#xff08;初始化&#xff09; 使用arange创造一个行向量&#xff0c;也就是0轴&#xff08;0维&a…
阅读更多...
LabVIEW 2023下载安装教程，附安装包和工具，免费使用，无套路获取

LabVIEW 2023下载安装教程，附安装包和工具，免费使用，无套路获取

前言 LabVIEW是一种程序开发环境&#xff0c;提供一种图形化编程方法&#xff0c;可可视化应用程序的各个方面&#xff0c;包括硬件配置、测量数据和调试&#xff0c;同时可以通过FPGA数学和分析选板中的NI浮点库链接访问浮点运算功能库&#xff0c;LabVIEW软件是NI设计平台的…
阅读更多...
西圣H1头戴式耳机发布，不止音质惊艳，更有舒适听音体验

西圣H1头戴式耳机发布，不止音质惊艳，更有舒适听音体验

近日&#xff0c;深耕智能声学领域多年的 xisem西圣品牌‏‏发布——‏西圣H1头戴式主动降噪蓝牙耳机正式迎来首销。从各个方面来看&#xff0c;这款头戴式耳机展现出了极具颠覆性的创新&#xff0c;不仅在音质和降噪性能上表现出色&#xff0c;更重要的是采用了更加轻巧的佩戴…
阅读更多...
使用记事本修复DBC文件问题V3.0

使用记事本修复DBC文件问题V3.0

要麻了&#xff0c;竟然写到3.0了。 有个信号解析不出来&#xff0c;同一条报文的其他信号可以正确解析&#xff0c;打开DBC检查&#xff0c;位置长度都是正确的&#xff0c;死活解析不出来。 打开记事本 发现他居然在信号后面加了个回车&#xff0c;就产生了上面这种诡异的情…
阅读更多...
js实现九九乘法表

js实现九九乘法表

效果图 代码 <!DOCTYPE html> <html><head><meta charset"utf-8"><title></title></head><body><script type"text/javascript">// 输出乘法口诀表// document.write () 空格 " " 换行…
阅读更多...
微电网优化MATLAB:火鹰优化算法（Fire Hawk Optimizer，FHO）求解微电网优化（提供MATLAB代码）

微电网优化MATLAB:火鹰优化算法（Fire Hawk Optimizer，FHO）求解微电网优化（提供MATLAB代码）

一、火鹰优化算法FHO 火鹰优化算法&#xff08;Fire Hawk Optimizer&#xff0c;FHO&#xff09;由Mahdi Azizi等人于2022年提出&#xff0c;该算法性能高效&#xff0c;思路新颖。 单目标优化&#xff1a;火鹰优化算法&#xff08;Fire Hawk Optimizer&#xff0c;FHO&#…
阅读更多...
QKCP容器平台安装qkcp paas deployment

QKCP容器平台安装qkcp paas deployment

22年底23年初做的容器云平台的项目&#xff0c;该份是当初实施部署真实生产环境的手册&#xff0c;关键ip和端口已经更改&#xff0c;现贴来给大伙参阅。 x公司qkcp容器平台安装部署手册 目录 x公司qkcp容器平台安装部署手册 1 集群基础环境 3 1.1. 集群规划信息 3 1.2. 标准…
阅读更多...
Eclipses安装教程

Eclipses安装教程

一、下载开发工具包 1、开发工具包JDK 下载地址链接&#xff1a;https://www.oracle.com/cn/java/technologies/downloads/ 下载教程&#xff1a; 1&#xff09;点击链接&#xff0c;可以跳转到页面 2&#xff09;下滑页面&#xff0c;找到开发工具包 3&#xff09; 记住下载之…
阅读更多...
数据库管理Navicat Premium 15

数据库管理Navicat Premium 15

Navicat Premium 15是一款强大的数据库管理和开发工具&#xff0c;支持多种数据库类型&#xff0c;包括MySQL、MariaDB、SQL Server、SQLite、Oracle和PostgreSQL等。它提供了数据可视化、高效的操作、数据同步与备份等功能&#xff0c;以及用户友好的界面。Navicat Premium 15…
阅读更多...
【数据库原理】（37）Web与数据库

【数据库原理】（37）Web与数据库

随着网络的高速发展和网络服务的日趋完善&#xff0c;网络上的信息量呈几何级数增长。为了有效地组织、存储、管理和使用网上的信息&#xff0c;数据库技术被广泛地应用于网络领域。特别是在Internet上&#xff0c;已建立了数以万计的网站&#xff0c;其中大中型网站的后台大多…
阅读更多...
2.4 网络层01

2.4 网络层01

2.4 网络层01 2.4.1 网络层概述 网络层的主要任务是实现网络互连&#xff0c;进而实现数据包在各网络之间的传输。 异构网络内部的计算机要想实现通信是不需要实现网络互联的&#xff0c;异构网络之间要想实现通信就必须实现网络互连。 路由器工作在五层协议体系结构的网络…
阅读更多...
Android：JNI实战，加载三方库、编译C/C++

Android：JNI实战，加载三方库、编译C/C++

一.概述 Android Jni机制让开发者可以在Java端调用到C/C&#xff0c;也是Android应用开发需要掌握的一项重要的基础技能。 计划分两篇博文讲述Jni实战开发。 本篇主要从项目架构上剖析一个Android App如何通过Jni机制加载三方库和C/C文件。 二.Native C Android Studio可…
阅读更多...
python一元线性回归sklearn

python一元线性回归sklearn

# -*- coding: utf-8 -*-from sklearn.linear_model import LinearRegression import numpy as np import matplotlib.pyplot as plt# 载入数据 data np.genfromtxt(一元线性回归.csv, delimiter,) x_data data[:,0] y_data data[:,1] plt.scatter(x_data,y_data) plt.show(…
阅读更多...
如何本地安装Python Flask并结合内网穿透实现远程开发

如何本地安装Python Flask并结合内网穿透实现远程开发

&#x1f49d;&#x1f49d;&#x1f49d;欢迎来到我的博客&#xff0c;很高兴能够在这里和您见面&#xff01;希望您在这里可以感受到一份轻松愉快的氛围&#xff0c;不仅可以获得有趣的内容和知识&#xff0c;也可以畅所欲言、分享您的想法和见解。 推荐:kwan 的首页,持续学…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023