Kafka 生产调优

Kafka生产调优

文章目录

  • Kafka生产调优
    • 一、Kafka 硬件配置选择
      • 场景说明
      • 服务器台数选择
      • 磁盘选择
      • 内存选择
      • CPU选择
    • 二、Kafka Broker调优
      • Broker 核心参数配置
      • 服役新节点/退役旧节点
      • 增加副本因子
      • 调整分区副本存储
    • 三、Kafka 生产者调优
      • 生产者如何提高吞吐量
      • 数据可靠性
      • 数据去重
      • 数据乱序
    • 四、Kafka 消费者调优
      • 消费者重要参数🚩
      • 消费者再平衡
      • 指定offset进行消费
      • 指定时间进行消费
      • 消费者如何提高吞吐量
    • 五、Kafka总体调优
      • 如何提升吞吐量🚩
      • 数据精准一次
      • 合理设置分区数
      • 单条日志大于 1m的问题
      • 集群压力测试
      • Kafka Producer 压力测试
      • Kafka Consumer 压力测试

一、Kafka 硬件配置选择

场景说明

100 万日活，每人每天 100 条日志，每天总共的日志条数是 100 万 * 100 条 = 1 亿条。

1 亿 / 24 小时 / 60 分 / 60 秒 = 1150 条/每秒钟。

每条日志大小：0.5k ~ 2k（约1k）。

1150 条/每秒钟 * 1k ≈ 1m/s 。

高峰期每秒钟：1150 条 * 20 倍 = 23000 条。

所以高峰每秒数据量：20MB/s。

服务器台数选择

服务器台数= 2 * （生产者峰值生产速率 * 副本 / 100） + 1

​ = 2 * （20m/s * 2 / 100） + 1

​ = 3 台

磁盘选择

kafka 底层主要是顺序写，固态硬盘和机械硬盘的顺序写速度差不多。建议选择普通的机械硬盘。

每天总数据量：1 亿条 * 1k ≈ 100g

100g * 副本 2 * 保存时间 3 天 / 0.7 ≈ 1T

建议三台服务器硬盘总大小，大于等于 1T。

内存选择

Kafka 内存组成：堆内存 + 页缓存

1）Kafka 堆内存建议每个节点：10g ~ 15g（在 kafka-server-start.sh 中修改）

if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
 export KAFKA_HEAP_OPTS="-Xmx10G -Xms10G"
fi

2）页缓存：页缓存是 Linux 系统服务器的内存。我们只需要保证 1 个 segment（1g）中25%的数据在内存中就好。

每个节点页缓存大小 =（分区数 * 1g * 25%）/ 节点数。例如 10 个分区，页缓存大小=（10 * 1g * 25%）/ 3 ≈ 1g

建议服务器内存大于等于 11G。

CPU选择

  • num.io.threads = 8 负责写磁盘的线程数，整个参数值要占总核数的 50%。
  • num.replica.fetchers = 1 副本拉取线程数，这个参数占总核数的 50%的 1/3。
  • num.network.threads = 3 数据传输线程数，这个参数占总核数的 50%的 2/3。

建议 32 个 cpu core。

二、Kafka Broker调优

Broker 核心参数配置

在这里插入图片描述

参数名称描述
replica.lag.time.max.msISR 中，如果 Follower 长时间未向 Leader 发送通信请求或同步数据，则该 Follower 将被踢出 ISR。该时间阈值，默认30s。
auto.leader.rebalance.enable默认是 true。 自动 Leader Partition 平衡。
leader.imbalance.per.broker.percentage默认是 10%。每个 broker 允许的不平衡的 leader 的比率。如果每个 broker 超过了这个值，控制器会触发 leader 的平衡。
leader.imbalance.check.interval.seconds默认值 300 秒。检查 leader 负载是否平衡的间隔时间。
log.segment.bytesKafka 中 log 日志是分成一块块存储的，此配置是指 log 日志划分成块的大小，默认值 1G。
log.index.interval.bytes默认 4kb，kafka 里面每当写入了 4kb 大小的日志（.log），然后就往 index 文件里面记录一个索引。
log.retention.hoursKafka 中数据保存的时间，默认 7 天。
log.retention.minutesKafka 中数据保存的时间，分钟级别，默认关闭。
log.retention.msKafka 中数据保存的时间，毫秒级别，默认关闭。
log.retention.check.interval.ms检查数据是否保存超时的间隔，默认是 5 分钟。
log.retention.bytes默认等于-1，表示无穷大（也表示关闭）。超过设置的所有日志总大小，删除最早的 segment。
log.cleanup.policy默认是 delete，表示所有数据启用删除策略；如果设置值为 compact，表示所有数据启用压缩策
num.io.threads默认是 8。负责写磁盘的线程数。整个参数值要占总核数的 50%。
num.replica.fetchers默认是 1。副本拉取线程数，这个参数占总核数的 50%的 1/3
num.network.threads默认是 3。数据传输线程数，这个参数占总核数的50%的 2/3 。
log.flush.interval.messages强制页缓存刷写到磁盘的条数，默认是 long 的最大值，9223372036854775807。一般不建议修改， 交给系统自己管理。
log.flush.interval.ms每隔多久，刷数据到磁盘，默认是 null。一般不建议修改，交给系统自己管理。

服役新节点/退役旧节点

（1）创建一个要均衡的主题。

$ vim topics-to-move.json 
{
   "topics": [
 		{"topic": "first"}
 	],
 	"version": 1
}

（2）生成一个负载均衡的计划。

$ bin/kafka-reassign-partitions.sh --bootstrap-server node102:9092 --topics-to-move-json-file topics-to-move.json --broker-list "0,1,2,3" --generate

（3）创建副本存储计划（所有副本存储在 broker0、broker1、broker2、broker3 中），由步骤2生成的👆

$ vim increase-replication-factor.json

（4）执行副本存储计划。

$ bin/kafka-reassign-partitions.sh --bootstrap-server node102:9092 --reassignment-json-file increase-replication-factor.json --execute

（5）验证副本存储计划。

$ bin/kafka-reassign-partitions.sh --bootstrap-server node102:9092 --reassignment-json-file increase-replication-factor.json --verify

增加副本因子

1）创建测试 topic（3分区、1副本）

$ bin/kafka-topics.sh --bootstrap-server node102:9092 --create --partitions 3 --replication-factor 1 --
topic four

2）手动增加副本存储（3分区、3副本），创建副本存储计划，所有副本都指定存储在 broker0、broker1、broker2 中

vim increase-replication-factor.json

{"version":1,"partitions":[
	{"topic":"four","partition":0,"replicas":[0,1,2]},
	{"topic":"four","partition":1,"replicas":[0,1,2]},
	{"topic":"four","partition":2,"replicas":[0,1,2]}
]}

3）执行副本存储计划。

$ bin/kafka-reassign-partitions.sh --bootstrap-server node102:9092 --reassignment-json-file increase-replication-factor.json --execute

调整分区副本存储

（1）创建副本存储计划（所有副本都指定存储在 broker0、broker1、broker2 中）。

 vim increase-replication-factor.json

{
	"version":1,
	"partitions":[{"topic":"three","partition":0,"replicas":[0,1]},
	{"topic":"three","partition":1,"replicas":[0,1]},
	{"topic":"three","partition":2,"replicas":[1,0]},
	{"topic":"three","partition":3,"replicas":[1,0]}
	]
}

（2）执行副本存储计划。

$ bin/kafka-reassign-partitions.sh --bootstrap-server node102:9092 --reassignment-json-file increase-replication-factor.json --execute

（3）验证副本存储计划

$ bin/kafka-reassign-partitions.sh --bootstrap-server node102:9092 --reassignment-json-file increase-replication-factor.json --verify

三、Kafka 生产者调优

生产者如何提高吞吐量

参数名称描述
buffer.memoryRecordAccumulator 缓冲区总大小，默认 32m。
batch.size缓冲区一批数据最大值，默认 16k。适当增加该值，可 缓冲区一批数据最大值，默认 16k。适当增加该值，可以提高吞吐量，但是如果该值设置太大，会导致数据传 以提高吞吐量，但是如果该值设置太大，会导致数据传输延迟增加。 输延迟增加。
linger.ms如果数据迟迟未达到 batch.size，sender 等待 linger.time之后就会发送数据。单位 ms，默认值是 0ms，表示没有延迟。生产环境建议该值大小为 5-100ms 之间。
compression.type生产者发送的所有数据的压缩方式。默认是 none，也就是不压缩。 支持压缩类型：none、gzip、snappy、lz4 和 zstd。

数据可靠性

参数名称描述
acks0：生产者发送过来的数据，不需要等数据落盘应答。
1：生产者发送过来的数据，Leader 收到数据后应答。
-1（all）：生产者发送过来的数据，Leader+和 isr 队列 里面的所有节点收齐数据后应答。默认值是-1

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

数据去重

参数名称描述
enable.idempotence是否开启幂等性，默认 true，表示开启幂等性。

Kafka 的事务一共有如下 5 个 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;

数据乱序

参数名称描述
enable.idempotence是否开启幂等性，默认 true，表示开启幂等性。
max.in.flight.requests.per.connection允许最多没有返回 ack 的次数，默认为 5，开启幂等性 要保证该值是 1-5 的数字。

四、Kafka 消费者调优

消费者重要参数🚩

参数描述
bootstrap.servers向 Kafka 集群建立初始连接用到的 host/port 列表。
key.deserializer 和 value.deserializer指定接收消息的 key 和 value 的反序列化类型。一定要写全类名。
group.id标记消费者所属的消费者组。
enable.auto.commit自动提交offset开关，默认值为 true，消费者会自动周期性地向服务器提交偏移量。
auto.commit.interval.ms消费者偏移量向Kafka提交的频率，默认5s。（如果设置自动提交offset时才生效）
auto.offset.reset当 Kafka 中没有初始偏移量或当前偏移量在服务器中不存在（如，数据被删除了），该如何处理？ earliest：自动重置偏移量到最早的偏移量。
latest：默认，自动重置偏移量为最新的偏移量。
none：如果消费组原来的（previous）偏移量
offsets.topic.num.partitions__consumer_offsets 的分区数，默认是 50 个分区。
heartbeat.interval.msKafka 消费者和 coordinator 之间的心跳时间，默认 3s。 该条目的值必须小于 session.timeout.ms ，也不应该高于session.timeout.ms 的 1/3。
☘️session.timeout.msKafka consumer 和 coordinator 之间连接超时时间，默认 45s。 超过该值，该消费者被移除，消费者组执行再平衡。
☘️max.poll.interval.ms消费者处理消息的最大时长，默认是 5 分钟。超过该值，该消费者被移除，消费者组执行再平衡。
🚩fetch.min.bytes消费者获取服务器端一批消息最小的字节数。 默认 1 个字节
🚩fetch.max.wait.ms默认 500ms。如果没有从服务器端获取到一批数据的最小字节数。该时间到，仍然会返回数据。
🚩fetch.max.bytes默认 Default: 52428800（50 m）。消费者获取服务器端一批消息最大的字节数。如果服务器端一批次的数据大于该值 （50m）仍然可以拉取回来这批数据，因此，这不是一个绝对最大值。一批次的大小受 message.max.bytes （broker config）or max.message.bytes （topic config）影响。
🚩max.poll.records一次 poll 拉取数据返回消息的最大条数，默认是 500 条。

消费者再平衡

参数名称描述
heartbeat.interval.msKafka 消费者和 coordinator 之间的心跳时间，默认 3s。 该条目的值必须小于 session.timeout.ms，也不应该高于 session.timeout.ms 的 1/3。
session.timeout.msKafka 消费者和 coordinator 之间连接超时时间，默认 45s。 超过该值，该消费者被移除，消费者组执行再平衡。
max.poll.interval.ms消费者处理消息的最大时长，默认是 5 分钟。超过该值，该消费者被移除，消费者组执行再平衡。
partition.assignment.strategy消费者分区分配策略 ， 默 认 策 略 是 Range + CooperativeSticky。Kafka 可以同时使用多个分区分配策略。 可以选择的策略包括：Range、RoundRobin、Sticky、CooperativeSticky

指定offset进行消费

public class CustomConsumerByHandSync {
   public static void main(String[] args) {
   // 1. 创建 kafka 消费者配置类
   Properties properties = new Properties();
   // 2. 添加配置参数
   // 添加连接
   properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
   // 配置序列化 必须
   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");
     
   // 配置消费者组 
   properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test");
     
   // 是否自动提交 offset
   properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
     
   //3. 创建 kafka 消费者
   KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
     
   // 2 订阅一个主题
   ArrayList<String> topics = new ArrayList<>();
   topics.add("first");
   kafkaConsumer.subscribe(topics);
   
   //🚩 获取消费者分区信息，并指定offset进行消费
   Set<TopicPartition> assignment= new HashSet<>();
 	  while (assignment.size() == 0) {
 		 kafkaConsumer.poll(Duration.ofSeconds(1));
 		 // 获取消费者分区分配信息（有了分区分配信息才能开始消费）
 		 assignment = kafkaConsumer.assignment();
       }
 	    
      // 遍历所有分区，并指定 offset 从 1700 的位置开始消费
	  for (TopicPartition tp: assignment) {
	  	kafkaConsumer.seek(tp, 1700);
	  }
     
     // 消费数据
	 while (true){
 		// 读取消息
 	    ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofSeconds(1));
       		// 输出消息
 			for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
 				System.out.println(consumerRecord.value());
 			}
 	}
     
  }
}

指定时间进行消费

在生产环境中，会遇到最近消费的几个小时数据异常，想重新按照时间消费。例如要求按照时间消费前一天的数据，怎么处理？

public class CustomConsumerByHandSync {
   public static void main(String[] args) {
   // 1. 创建 kafka 消费者配置类
   Properties properties = new Properties();
   // 2. 添加配置参数
   // 添加连接
   properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
   // 配置序列化 必须
   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");
     
   // 配置消费者组 
   properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test");
     
   // 是否自动提交 offset
   properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
     
   //3. 创建 kafka 消费者
   KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
     
   // 2 订阅一个主题
   ArrayList<String> topics = new ArrayList<>();
   topics.add("first");
   kafkaConsumer.subscribe(topics);
   
   //🚩 获取消费者分区信息，并指定offset进行消费
   Set<TopicPartition> assignment= new HashSet<>();
   while (assignment.size() == 0) {
 	  kafkaConsumer.poll(Duration.ofSeconds(1));
 	  // 获取消费者分区分配信息（有了分区分配信息才能开始消费）
 	  assignment = kafkaConsumer.assignment();
   }
     
   HashMap<TopicPartition, Long> timestampToSearch = new HashMap<>();	    
   // 封装集合存储，每个分区对应一天前的数据
   for (TopicPartition topicPartition : assignment) {
       timestampToSearch.put(topicPartition, System.currentTimeMillis() - 1 * 24 * 3600 * 1000);
   }
    // 获取从 1 天前开始消费的每个分区的 offset
 	Map<TopicPartition, OffsetAndTimestamp> offsets = kafkaConsumer.offsetsForTimes(timestampToSearch);
    // 遍历每个分区，对每个分区设置消费时间。
 	for (TopicPartition topicPartition : assignment) {
 		OffsetAndTimestamp offsetAndTimestamp = offsets.get(topicPartition);
 		// 根据时间指定开始消费的位置
 		if (offsetAndTimestamp != null){
 			kafkaConsumer.seek(topicPartition, offsetAndTimestamp.offset());
 		}
 	}
     
     // 消费数据
	 while (true){
 		// 读取消息
 	    ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofSeconds(1));
       		// 输出消息
 			for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
 				System.out.println(consumerRecord.value());
 			}
 	}
     
  }
}

消费者如何提高吞吐量

增加分区数：

$ bin/kafka-topics.sh --bootstrap-server node102:9092 --alter --topic first --partitions 3
参数名称描述
fetch.max.bytes默认 Default: 52428800（50 m）。消费者获取服务器端一批 消息最大的字节数。如果服务器端一批次的数据大于该值 （50m）仍然可以拉取回来这批数据，因此，这不是一个绝对最大值。一批次的大小受 message.max.bytes （broker config） or max.message.bytes （topic config）影响。
max.poll.records一次 poll 拉取数据返回消息的最大条数，默认是 500 条。

五、Kafka总体调优

如何提升吞吐量🚩

1）提升生产吞吐量

  • buffer.memory：发送消息的缓冲区大小，默认值是 32m，可以增加到 64m。
  • batch.size：默认是 16k。如果 batch 设置太小，会导致频繁网络请求，吞吐量下降；如果 batch 太大，会导致一条消息需要等待很久才能被发送出去，增加网络延时
  • linger.ms，这个值默认是 0，意思就是消息必须立即被发送。一般设置一个 5~100毫秒。如果 linger.ms 设置的太小，会导致频繁网络请求，吞吐量下降；如果 linger.ms 太长，会导致一条消息需要等待很久才能被发送出去，增加网络延时。
  • compression.type：默认是 none，不压缩，但是也可以使用 lz4 压缩，效率还是不错的，压缩之后可以减小数据量，提升吞吐量，但是会加大 producer 端的 CPU 开销。

2）增加分区

3）消费者提高吞吐量

  • 调整 fetch.max.bytes 大小，默认是 50m。
  • 调整 max.poll.records 大小，默认是 500 条。

4）增加下游消费者处理能力

数据精准一次

1）生产者角度

  • acks 设置为-1 （acks=-1）
  • 幂等性（enable.idempotence = true） + 事务

2）broker 服务端角度

  • 分区副本大于等于 2 （–replication-factor 2）
  • ISR 里应答的最小副本数量大于等于 2 （min.insync.replicas = 2）

3）消费者

  • 事务 + 手动提交 offset （enable.auto.commit = false）
  • 消费者输出的目的地必须支持事务（MySQL、Kafka）

合理设置分区数

（1）创建一个只有 1 个分区的 topic。

（2）测试这个 topic 的 producer 吞吐量和 consumer 吞吐量。

（3）假设他们的值分别是 Tp 和 Tc，单位可以是 MB/s。

（4）然后假设总的目标吞吐量是 Tt，那么分区数 = Tt / min（Tp，Tc）。

例如：producer 吞吐量 = 20m/s；consumer 吞吐量 = 50m/s，期望吞吐量 100m/s；

分区数 = 100 / 20 = 5 分区

分区数一般设置为：3-10 个

分区数不是越多越好，也不是越少越好，需要搭建完集群，进行压测，再灵活调整分区个数。

单条日志大于 1m的问题

参数名称描述
message.max.bytes默认 1m，broker 端接收每个批次消息最大值
max.request.size默认 1m，生产者发往 broker 每个请求消息最大值。针对 topic级别设置消息体的大小
replica.fetch.max.bytes默认 1m，副本同步数据，每个批次消息最大值
fetch.max.bytes默认 Default: 52428800（50 m）。消费者获取服务器端一批 消息最大的字节数。如果服务器端一批次的数据大于该值（50m）仍然可以拉取回来这批数据，因此，这不是一个绝对 最大值。一批次的大小受 message.max.bytes （broker config） or max.message.bytes （topic config）影响

集群压力测试

用 Kafka 官方自带的脚本，对 Kafka 进行压测

  • 生产者压测：kafka-producer-perf-test.sh
  • 消费者压测：kafka-consumer-perf-test.sh

Kafka Producer 压力测试

（1）创建一个 test topic，设置为 3 个分区 3 个副本

$ bin/kafka-topics.sh --bootstrap-server node102:9092 --create --replication-factor 3 --partitions 3 --topic test

（2）在/opt/module/kafka/bin 目录下面有这两个文件。我们来测试一下

$ bin/kafka-producer-perf-test.sh --topic test --record-size 1024 --num-records 1000000 --throughput 10000 --producer-props bootstrap.servers=node102:9092,node103:9092,node104:9092 batch.size=16384 linger.ms=0

测试参数说明：

  • record-size 是一条信息有多大，单位是字节，本次测试设置为 1k。
  • num-records 是总共发送多少条信息，本次测试设置为 100 万条。
  • throughput 是每秒多少条信息，设成-1，表示不限流，尽可能快的生产数据，可测出生产者最大吞吐量。本次设置为每秒钟 1 万条。
  • producer-props 后面可以配置生产者相关参数，batch.size 配置为 16k

调优参数：

更改不同的值进行压测

参数名称描述
buffer.memoryRecordAccumulator 缓冲区总大小，默认 32m。
batch.size缓冲区一批数据最大值，默认 16k。适当增加该值，可 缓冲区一批数据最大值，默认 16k。适当增加该值，可以提高吞吐量，但是如果该值设置太大，会导致数据传 以提高吞吐量，但是如果该值设置太大，会导致数据传输延迟增加。 输延迟增加。
linger.ms如果数据迟迟未达到 batch.size，sender 等待 linger.time之后就会发送数据。单位 ms，默认值是 0ms，表示没有延迟。生产环境建议该值大小为 5-100ms 之间。
compression.type生产者发送的所有数据的压缩方式。默认是 none，也就是不压缩。 支持压缩类型：none、gzip、snappy、lz4 和 zstd。

Kafka Consumer 压力测试

（1）修改/opt/module/kafka/config/consumer.properties 文件中的一次拉取条数为 500

max.poll.records=500

（2）消费 100 万条日志进行压测

$ bin/kafka-consumer-perf-test.sh --bootstrap-server node102:9092,node103:9092,hadoop104:9092 --topic test --messages 1000000 --consumer.config config/consumer.properties

参数说明：

  • –bootstrap-server 指定 Kafka 集群地址
  • –topic 指定 topic 的名称
  • –messages 总共要消费的消息个数。本次实验 100 万条

调优参数优化：（在consumer.properties中进行修改）

  • 修改一次拉取的数量：max.poll.records=2000
  • 调整文件中的拉取一批数据大小 100m：fetch.max.bytes=104857600

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

如若内容造成侵权/违法违规/事实不符，请联系我们进行投诉反馈qq邮箱809451989@qq.com，一经查实，立即删除！

相关文章

智慧城市：打造低碳未来，引领城市数字化转型新篇章

智慧城市：打造低碳未来，引领城市数字化转型新篇章

在“万物皆可数字化”的新时代浪潮下&#xff0c;智慧城市作为未来城市发展的先锋方向&#xff0c;正在以前所未有的速度和规模重塑我们的城市面貌。 智慧城市不仅是一个技术革新的标志&#xff0c;更是城市治理、民生服务等领域全面升级的重要引擎。 一、智慧城市的多元应用领…
阅读更多...
Bootstrap5 导航组件和面包屑

Bootstrap5 导航组件和面包屑

Bootstrap5 导航组件和面包屑 Bootstrap5 提供了一种简单快捷的方法来创建基本导航&#xff0c;它提供了非常灵活和优雅的选项卡和Pills等组件。 Bootstrap5 的所有导航组件&#xff0c;包括选项卡和Pills&#xff0c;都通过基本的 .nav 类共享相同的基本标记和样式。 使用 B…
阅读更多...
2024Node.js零基础教程(小白友好型)，nodejs新手到高手，（六）NodeJS入门——http模块

2024Node.js零基础教程(小白友好型)，nodejs新手到高手，（六）NodeJS入门——http模块

047_http模块_获取请求行和请求头 hello&#xff0c;大家好&#xff0c;那第二节我们来介绍一下如何在这个服务当中来提取 HTT 请求报文的相关内容。首先先说一下关于报文的提取的方法&#xff0c;我在这个文档当中都已经记录好了&#xff0c;方便大家后续做一个快速的查阅。 …
阅读更多...
springboot集成easypoi导出多sheet页

springboot集成easypoi导出多sheet页

pom文件 <dependency><groupId>cn.afterturn</groupId><artifactId>easypoi-base</artifactId><version>4.1.0</version> </dependency> 导出模板&#xff1a; 后端代码示例&#xff1a; /*** 导出加油卡进便利店大额审批列…
阅读更多...
洛谷_P1464 Function_python写法

洛谷_P1464 Function_python写法

目录 1.错误解法 2.学习记忆化搜索算法 2.1简介 2.2案例学习 3.解法 4.总结 1.错误解法 a 0 b 0 c 0 def w(a,b,c):if a<0 or b<0 or c<0:return 1elif a>20 or b>20 or c>20:return w(20,20,20)elif a<b and b<c:return w(a-1,b,c) w(a-1,…
阅读更多...
Python程序员面试题精选及解析（2）

Python程序员面试题精选及解析（2）

本文精心挑选了10道Python程序员面试题&#xff0c;覆盖了Python的多个核心领域&#xff0c;包括装饰器、lambda函数、列表推导式、生成器、全局解释器锁(GIL)、单例模式以及上下文管理器等。每道题都附有简洁的代码示例&#xff0c;帮助读者更好地理解和应用相关知识点无论是对…
阅读更多...
小白代码审计入门

小白代码审计入门

最近小白一直在学习代码审计，对于我这个没有代码审计的菜鸟来说确实是一件无比艰难的事情。但是着恰恰应了一句老话：万事开头难。但是小白我会坚持下去。何况现在已经喜欢上了代码审计，下面呢小白就说一下appcms后台模板Getshell以及读取任意文件，影响的版本是2.0.101版本。…
阅读更多...
【数据分享】1929-2023年全球站点的逐月平均降水量（Shp\Excel\免费获取）

【数据分享】1929-2023年全球站点的逐月平均降水量（Shp\Excel\免费获取）

气象数据是在各项研究中都经常使用的数据&#xff0c;气象指标包括气温、风速、降水、湿度等指标&#xff0c;说到常用的降水数据&#xff0c;最详细的降水数据是具体到气象监测站点的降水数据&#xff01; 有关气象指标的监测站点数据&#xff0c;之前我们分享过1929-2023年全…
阅读更多...
基于spring cloud alibaba的微服务平台架构规划

基于spring cloud alibaba的微服务平台架构规划

平台基础能力规划&#xff08;继续完善更新…&#xff09; 一、统一网关服务&#xff08;独立服务&#xff09; 二、统一登录鉴权系统管理&#xff08;独立服务&#xff09; 1.统一登录 2.统一鉴权 3.身份管理 用户管理 角色管理 业务系统和菜单管理 部门管理 岗位管理 字典管…
阅读更多...
【代码随想录24】93.复原 IP 地址 78.子集 90.子集II

【代码随想录24】93.复原 IP 地址 78.子集 90.子集II

目录 93.复原IP地址题目描述参考代码 78.子集题目描述参考代码 90.子集II题目描述参考代码 93.复原IP地址 题目描述 有效 IP 地址 正好由四个整数&#xff08;每个整数位于 0 到 255 之间组成&#xff0c;且不能含有前导 0&#xff09;&#xff0c;整数之间用 . 分隔。 例如…
阅读更多...
Java设计模式大全：23种常见的设计模式详解（三）

Java设计模式大全：23种常见的设计模式详解（三）

本系列文章简介&#xff1a; 设计模式是在软件开发过程中&#xff0c;经过实践和总结得到的一套解决特定问题的可复用的模板。它是一种在特定情境中经过验证的经验和技巧的集合&#xff0c;可以帮助开发人员设计出高效、可维护、可扩展和可复用的软件系统。设计模式提供了一种在…
阅读更多...
VitePress-11-静态资源引入- public目录的使用

VitePress-11-静态资源引入- public目录的使用

作用简述 public目录就是一个存放静态资源的目录。 项目中可以通过【绝对路径】的方式进行引入。 具体的格式请继续阅读本文剩下的内容。 知识补充-源目录(比较重要) 源目录 的概念 &#xff1a; 是markdown源文件所在的位置。 默认情况下&#xff0c;源目录与项目的根目录保持…
阅读更多...
警惕“中等数字化陷阱”，大力发扬先进基础设施“长板”

警惕“中等数字化陷阱”，大力发扬先进基础设施“长板”

上世纪七、八十年代&#xff0c;拉美国家发展由富转穷&#xff0c;人均GDP发展至3000美金左右就开始停滞不前。研究界将这一现象归结为一个极具争议的概念——“中等收入陷阱”。 如今&#xff0c;在我国数字化发展当中&#xff0c;也有一种“中等数字化陷阱”正露出苗头&…
阅读更多...
读千脑智能笔记06_人工智能的未来（上）

读千脑智能笔记06_人工智能的未来（上）

1. 人工智能正在经历一场复兴&#xff0c;这是科技界最热门的领域之一 1.1. 大多数科学进步都建立在被广泛接受的理论框架之上。这种理论框架称为“科学范式” 1.2. 人工智能的未来将与大多数人工智能从业者如今的设想存在本质上的不同 1.3. 人工神经网络与人类大脑中的神经…
阅读更多...
Swift Combine 从入门到精通一

Swift Combine 从入门到精通一

1. Combine 简介 用 Apple 官方的话来说&#xff0c;Combine 是: a declarative Swift API for processing values over time. Combine 是 Apple 用来实现函数响应式编程的库&#xff0c; 类似于 RxSwift。 RxSwift 是 ReactiveX 对 Swift 语言的实现。 Combine 使用了许多可以…
阅读更多...
uniapp canvas游标卡尺效果

uniapp canvas游标卡尺效果

效果 根据公司业务仿照写的效果。原项目从微信小程序转uniapp，未测试该效果在android端效果。 uniapp直接使用canvas不可做子组件，否则无效果显示，其次显示时要考虑页面渲染超时的问题。 如效果所见，可以设置取值精度。 gitee地址：project_practice: 项目练习 - Gitee.…
阅读更多...
Qt/C++音视频开发66-音频变速不变调/重采样/提高音量/变速变调/倍速播放/sonic库使用

Qt/C++音视频开发66-音频变速不变调/重采样/提高音量/变速变调/倍速播放/sonic库使用

一、前言 之前在做倍速这个功能的时候&#xff0c;发现快速播放会有滴滴滴的破音出现&#xff0c;正常1倍速没有这个问题&#xff0c;尽管这个破音间隔很短&#xff0c;要放大音量才能听到&#xff0c;但是总归是不完美的&#xff0c;后面发现&#xff0c;通过修改qaudiooutpu…
阅读更多...
用python编写爬虫，爬取二手车信息+实验报告

用python编写爬虫，爬取二手车信息+实验报告

题目 报告要求 工程报告链接放在这里 https://download.csdn.net/download/Samature/88805518使用 1.安装jupyter notebook 2.用jupyter notebook打开工程里的ipynb文件&#xff0c;再run all就行 注意事项 可能遇到的bug 暂无&#xff0c;有的话私信我
阅读更多...
跟着pink老师前端入门教程-day20

跟着pink老师前端入门教程-day20

二、移动WEB开发之flex布局 1、flex 布局体验 1.1 传统布局与flex布局 传统布局&#xff1a;兼容性好、布局繁琐、局限性、不能再移动端很好的布局 flex弹性布局&#xff1a;操作方便&#xff0c;布局极为简单&#xff0c;移动端应用很广泛&#xff1b;PC 端浏览器支持情况…
阅读更多...
【HTML】MDN

【HTML】MDN

文章目录 一、html元素1.1 <a>1.2 <abbr>1.3 <address>1.4<area>1.5 <article>1.6 <aside>1.7 <audio>1.8 <b>1.9 <base>1.10<bdi>1.11 <bdo>1.12 <blockquote>1.13 <body>1.14 <br>1.15…
阅读更多...
最新文章

Copyright @ 2022~2023

友情链接： 我的编程经验分享 一条长河网 尧图网站开发 尧图建网站