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" ]; thenexport 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

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

2)手动增加副本存储,创建副本存储计划(所有副本都指定存储在 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");// 是否自动提交 offsetproperties.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");// 是否自动提交 offsetproperties.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 天前开始消费的每个分区的 offsetMap<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.mzph.cn/news/670188.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

算法day11

算法day11 239 滑动窗口最大值237 前K个高频元素栈与队列总结 滑动窗口最大值 第一想法&#xff0c;暴力解&#xff1a;这个解法会超时。&#xff08;这就是为啥是困难题&#xff09; 思路&#xff1a;每到一个新的窗口&#xff0c;就重新进行一次窗口中的max迭代&#xff0c…

ubuntu22.04@laptop OpenCV Get Started: 000_hello_opencv

ubuntu22.04laptop OpenCV Get Started: 000_hello_opencv 1. 源由2. Hello OpenCV2.1 C应用Demo2.2 Python应用Demo 3. 参考资料 1. 源由 之前&#xff0c;通过敲门砖已经砸开了OpenCV的大门&#xff0c;接下来是体验下“Hello World&#xff01;”程序。 2. Hello OpenCV …

HarmonyOS鸿蒙ArkTS证件照生成模板(适合二次开发,全套源码版)

预览效果 部分代码 开发语言 HarmonyOS 鸿蒙 ArkTS语言 &#xff08;Stage模型&#xff09; 备注 一键生成&#xff0c;自带证件照数集&#xff0c; 为开发者带来二次开发和学习体验&#xff0c; 在这祝福开发者们使用愉快。 使用方法 下载后通过DevEco Studio开发工…

物联网ARM开发-STM32之RTC浅谈

RTC 一.RTC简单介绍 RTC好比我们用来记录时间的一个钟表&#xff0c;他里面有年月日&#xff0c;还可以记录星期&#xff0c;小时&#xff0c;分钟等。是Real Time Clock的缩写&#xff0c;译为实时时钟&#xff0c;本质上是一个独立的定时器。 1. 1 与通用定时器的区别 可以…

EEPROM之MB85RC64介绍

一、芯片介绍 工作频率 : 400 kHz &#xff08;最大&#xff09; 即&#xff0c;当主机和一个EEPROM通信时&#xff0c;从机地址为1010 000WR&#xff0c;如果主机和多个EEPROM通信时&#xff0c;从机地址为1010 A2 A1 A0 WR 二、时序分析 &#xff08;1&#xff09;从机地址…

适用于 Windows 和 Mac 的 16 款最佳数据恢复软件

数据恢复软件是找回因硬盘损坏、病毒攻击或意外删除数据等原因而在设备上丢失的数据的最佳方法。在数字世界中&#xff0c;丢失数据是一件非常糟糕的事情&#xff0c;这会让许多人的情况变得更糟。使用最佳数据恢复软件可以减轻您必须努力恢复丢失数据的压力。它将带回您的大部…

C# OpenCvSharp DNN 部署yolov4目标检测

目录 效果 项目 代码 下载 效果 项目 代码 using OpenCvSharp; using OpenCvSharp.Dnn; using System; using System.Collections.Generic; using System.Drawing; using System.IO; using System.Linq; using System.Windows.Forms;namespace OpenCvSharp_DNN_Demo {publ…

Linux环境下配置mysql主从复制

主从配置需要注意的地方 1、主DB server和从DB server数据库的版本一致 2、主DB server和从DB server数据库数据一致[这里就会可以把主的备份在从上还原&#xff0c;也可以直接将主的数据目录拷贝到从的相应数据目录] 3、主DB server开启二进制日志,主DB server和从DB serve…

二分查找------蓝桥杯

题目描述&#xff1a; 请实现无重复数字的升序数组的二分查找 给定一个元素升序的、无重复数字的整型数组 nums 和一个目标值 target&#xff0c;写一个函数搜索 nums 中的target&#xff0c;如果目标值存在返回下标 (下标从0 开始)&#xff0c;否则返回-1 数据范围: 0 < l…

分布式文件存储系统minio

参考Linux搭建免费开源对象存储 wget https://dl.minio.io/server/minio/release/linux-amd64/minio yum install -y wget yum install -y wget wget https://dl.minio.io/server/minio/release/linux-amd64/minio chmod x minio sudo mv minio /usr/local/bin/ minio --vers…

外汇天眼:外汇天眼:注意,19个外汇平台因诈骗被监管拉黑!

近年来&#xff0c;全球金融市场出现了众多非法投资平台&#xff0c;这些平台利用虚假宣传和高回报承诺欺骗投资者&#xff0c;造成了严重的经济损失。为了保护投资者利益&#xff0c;监管机构也在加大力度打击这些非法平台。就在最近&#xff0c;又有19个外汇交易平台因涉嫌诈…

【C语言 - 力扣 - 反转链表】

反转链表题目描述 给你单链表的头节点 head &#xff0c;请你反转链表&#xff0c;并返回反转后的链表。 题解1-迭代 假设链表为 1→2→3→∅&#xff0c;我们想要把它改成 ∅←1←2←3。 在遍历链表时&#xff0c;将当前节点的 next 指针改为指向前一个节点。由于节点没…

计算机网络-差错控制(纠错编码 海明码 纠错方法)

文章目录 纠错编码-海明码海明距离1.确定校验码位数r2.确定校验码和数据的位置3.求出校验码的值4.检错并纠错纠错方法1纠错方法2 小结 纠错编码-海明码 奇偶校验码&#xff1a;只能发现错误不能找到错误位置和纠正错误 海明距离 如果找到码距为1&#xff0c;那肯定为1了&…

idea运行程序报错 java 程序包org.junit不存在

在 IntelliJ IDEA 中运行程序时遇到错误提示&#xff1a;“java: 程序包org.junit不存在”&#xff0c;针对这一问题&#xff0c;我们可以考虑以下三步来解决&#xff1a; 第一步&#xff1a;检查JUnit依赖 尽管现代项目创建时通常会默认引入JUnit依赖&#xff0c;但仍需检查…

双向链表的插入、删除、按位置增删改查、栈和队列区别、什么是内存泄漏

2024年2月4日 1.请编程实现双向链表的头插&#xff0c;头删、尾插、尾删 头文件&#xff1a; #ifndef __HEAD_H__ #define __HEAD_H__ #include<stdio.h> #include<stdlib.h> #include<string.h> typedef int datatype; enum{FALSE-1,SUCCSE}; typedef str…

2、ChatGPT 在数据科学中的应用

ChatGPT 在数据科学中的应用 ChatGPT 可以成为数据科学家的绝佳工具。以下是我所了解到的关于它擅长的地方和不那么擅长的地方。 我从使用 ChatGPT 中学到了一个教训。它在数据科学中非常有帮助,但你必须仔细检查它输出的所有内容。它非常适合某些任务,并且可以非常快速准确…

Visual Studio 20XX控制台程序鼠标点击阻塞问题

文章目录 方法一方法二 在Visual Studio 20xx编写的控制台程序中&#xff0c;当鼠标点击控制台时&#xff0c;会阻塞控制台程序运行&#xff0c;不按回车无法继续运行。 方法一 右击控制台标题栏&#xff0c;选择属性&#xff0c;去掉快速编辑模式(Q)的勾选&#xff0c;如&…

进程状态、排队

进程状态 1. 进程排队2. 进程各个状态3. 查看一个进程的状态4. linux内核描述进程状态5. 孤儿进程 1. 进程排队 进程为什么要排队呢&#xff1f;答案就是资源不够。需要等待某个软硬件资源&#xff0c;就像我们常用的scanf函数就是等待键盘资源。 之前的文章我们有个结论&#…

java面试题:MySQL中的各种JOIN的区别

表关联是频率非常高的一种数据库操作&#xff0c;在MySQL中&#xff0c;这种JOIN操作有很多类型&#xff0c;包括内联接、左外连接、右外连接等等&#xff0c;而每种连接的含义都不一样&#xff0c;如果死记硬背&#xff0c;不仅很难记住&#xff0c;而且也容易搞混淆&#xff…

pytorch 利用Tensorboar记录训练过程loss变化

文章目录 1. LossHistory日志类定义2. LossHistory类的使用2.1 实例化LossHistory2.2 记录每个epoch的loss2.3 训练结束close掉SummaryWriter 3. 利用Tensorboard 可视化3.1 显示可视化效果 参考 利用Tensorboard记录训练过程中每个epoch的训练loss以及验证loss&#xff0c;便于…