「Kafka」生产者篇

「Kafka」生产者篇

生产者发送消息流程

在消息发送的过程中，涉及到了 两个线程 ——main 线程和Sender 线程。

在 main 线程中创建了 一个 双端队列 RecordAccumulator。

main线程将消息发送给RecordAccumulator，Sender线程不断从 RecordAccumulator 中拉取消息发送到 Kafka Broker。

image-20231222155757635

  • main线程创建 Producer 对象，调用 send 函数发送消息，经过：
    • 拦截器 Interceptors（可选项，扩展一些额外功能）
    • 序列化器 Serializer（为什么不用Java的序列化？因为大数据传输需要更轻量的序列化方式）
    • 分区器 Partitioner，需要判断发送到哪个分区
  • 一个分区就会创建一个双端队列 RecordAccumulator，创建队列都是在内存里完成的，总大小默认为 32m。
    • 双端队列 RecordAccumulator 还有一个内存池的概念，每次 send 数据到队列后，在存放数据的时候会从内存池中取出内存，数据发送到kafka后释放内存归还到内存池；一端创建内存，另一端释放内存，这也是它为什么设计为双端队列。
  • Sender线程从队列中拉取数据
    • 每次批处理batch.size的大小默认为 16k，延迟时间 linger.ms 默认为 0ms，没有延迟。
      • 这两个条件是 或 的关系，两个条件达到任意一个就可以发送数据。
    • 以节点的方式， key:value => Broker1:(队列数据...) 的格式发送给对应的 kafka 服务器，如果kafka没有应答，默认每个broker节点队列最多缓存 5 个请求，后续 生产经验—数据乱序 的章节会讲这个作用。
  • Selector负责打通底层的链路，IO输入流 => IO输出流，经过Selector发送到kafka集群，kafka集群进行副本的同步。
  • 如果kafka集群收到数据后，会返回 ack，有3种模式，如上图。
    • 如果ack返回成功，则先清理掉缓存的Request请求，然后清理到对应队列中的数据。
    • 如果ack返回失败，则进行 retries 重试，默认重试次数是int的最大值（死磕），一直发Request请求，直到重试成功。
    • 详细讲解请参考下文的 生产经验—数据可靠性。

生产者重要参数列表

image-20231222171445101

image-20231222171536700

image-20231222171611680

异步发送

  • 同步发送：外部数据发送到 RecordAccumulator 队列中，等待这批数据都发送到 kafka 集群，再返回。
  • 异步发送：外部数据发送到 RecordAccumulator 队列中，不管这些数据有没有发送到 kafka 集群，直接返回。
    • 默认为异步发送

普通异步发送

编写不带回调函数的代码

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 配置对象添加配置信息：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");
		
		// 3. 创建 kafka 生产者对象
		KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
		
		// 4. 调用 send 方法,发送消息
		for (int i = 0; i < 5; i++) {
            // 这里只指定了topic和value
            kafkaProducer.send(new ProducerRecord<>("first", "atguigu " + i));
		}
		
		// 5. 关闭资源
		kafkaProducer.close();
	}
}

回调异步发送

回调函数会在 producer 收到 ack 时调用，为异步调用，该方法有两个参数，分别是元数据信息（RecordMetadata）和异常信息（Exception）。

如果 Exception 为 null，说明消息发送成功，如果 Exception 不为 null，说明消息发送失败。

注意：消息发送失败会自动重试，不需要我们在回调函数中手动重试。

image-20231222165627441

// 4. 调用 send 方法,发送消息
for (int i = 0; i < 5; i++) {
    // 添加回调 Callback
    kafkaProducer.send(new ProducerRecord<>("first", "atguigu " + i), new Callback() {
        // 该方法在 Producer 收到 ack 时调用，为异步调用
        @Override
        public void onCompletion(RecordMetadata metadata, Exception exception) {
            if (exception == null) {
                // 没有异常，输出信息到控制台
                System.out.println("主题：" + metadata.topic() + "->" +
                                "分区：" + metadata.partition());
            } else {
                // 出现异常打印
                exception.printStackTrace();
            }
        }
    });
    // 延迟一会会看到数据发往不同分区
    Thread.sleep(2);
}

同步发送

只需在异步发送的基础上，再调用一下 get() 方法即可。

image-20231222172334948

生产者分区

分区好处

image-20231222172447587

可以通过机器的存储能力自定义分区数据，比如 broker0 存储 20T 数据，broker1和2分别存储 40T 数据。

生产者发送消息的分区策略

可阅读：详解Kafka分区机制原理｜Kafka 系列 二

默认的分区器 DefaultPartitioner

/**
 * The default partitioning strategy: 默认分区策略
 * 如果你指定了分区，则直接用这个分区
 * 如果没指定分区，但有key，则按照key的hash值 % 分区数
 * 如果既没指定分区也没指定key，则按照粘性分区处理。
 * See KIP-480 for details about sticky partitioning.
 */
public class DefaultPartitioner implements Partitioner {
	...
}

ProducerRecord 类的构造方法就表示了这 3 种分区策略：

image-20231222173831306

自定义分区器

  • 定义类实现 Partitioner 接口

  • 重写 partition() 方法

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

/**
 * 1. 实现接口 Partitioner
 * 2. 实现3个方法: partition、close、configure
 * 3. 编写 partition 方法，返回分区号
 */
public class MyPartitioner implements Partitioner {
	/**
	 * 返回信息对应的分区
	 * @param topic 主题
	 * @param key 消息的 key
	 * @param keyBytes 消息的 key 序列化后的字节数组
	 * @param value 消息的 value
	 * @param valueBytes 消息的 value 序列化后的字节数组
	 * @param cluster 集群元数据可以查看分区信息
	 * @return
	 */
	@Override
	public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
		// 获取消息
		String msgValue = value.toString();
		// 创建 partition
		int partition;
		// 判断消息是否包含 atguigu
		if (msgValue.contains("atguigu")) {
			partition = 0;
		} else {
			partition = 1;
		}
		// 返回分区号
		return partition;
	}
	
	// 关闭资源
	@Override
	public void close() {
	}
	
	// 配置方法
	@Override
	public void configure(Map<String, ?> configs) {
	}
}

  • 使用分区器的方法，在生产者的配置中添加分区器参数

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

public class CustomProducerCallbackPartitions {
    public static void main(String[] args) {
        Properties properties = new Properties();
        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());

        // 添加自定义分区器
        properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, "com.atguigu.kafka.producer.MyPartitioner");
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);

        for (int i = 0; i < 5; i++) {
            kafkaProducer.send(new ProducerRecord<>("first", "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();
    }
}

生产者如何提高吞吐量

  1. 合理调整 batch.size 和 linger.ms 的参数值
  2. 采用数据压缩
  3. 调整缓冲区大小

image-20231222181126685

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 CustomProducerParameters {
    public static void main(String[] args) {
        // 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());
        // batch.size：批次大小，默认 16K
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
        // linger.ms：等待时间，默认 0ms
        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<>(properties);
        
        // 4. 调用 send 方法,发送消息
        for (int i = 0; i < 5; i++) {
            kafkaProducer.send(new ProducerRecord<>("first", "atguigu " + i));
        }
        
        // 5. 关闭资源
        kafkaProducer.close();
    }
}

生产经验—数据可靠性

回顾发送流程

image-20231227174242122

数据可靠性主要根据 kafka 集群返回给我们的 ack。

ack 应答原理

  • ack=0，不需要等待数据落盘应答，一直发送给 kafka，很容易丢数据。
    • 数据发送到 Leader 后，Leader 挂掉了，此时数据还在内存中，未落盘，数据丢失。
  • ack=1，不需要等待 kafka 主从同步完成，Leader 收到数据落盘后应答。
    • Leader 成功落盘，但还未同步给 Follower，Leader 挂了，数据丢失。
  • ack=-1，需要等待 Leader 和 ISR 队列里面的所有节点收齐数据后应答。

image-20231227174844019

数据完全可靠条件

image-20231227174931962

数据完全可靠条件 = ACK 级别设置为-1 + 分区副本大于等于2 + ISR里应答的最小副本数量大于等于2

注意，这里的“副本”并不是指的 Follower；在 Kafka 中，副本分为 Leader 副本和 Follower 副本。Leader 副本负责处理消息，而 Follower 副本则简单地复制 Leader 副本的数据。

也就是一个分区至少要有 1 个 Leader 和 1 个 Follower，ISR 队列最少也要有 1 个 Leader 和 1 个 Follower。

一个分区至少有 1 个 Leader，所以每个 Partition 都会有一个 ISR，而且是由 Leader 动态维护。

可靠性总结

  • acks=0，生产者发送过来数据就不管了，可靠性差，效率高；
  • acks=1，生产者发送过来数据 Leader 应答，可靠性中等，效率中等；
  • acks=-1，生产者发送过来数据 Leader 和 ISR 队列里面所有 Follwer 应答，可靠性高，效率低；
  • 在生产环境中，
    • acks=0，很少使用；
    • acks=1，一般用于传输普通日志，允许丢个别数据；
    • acks=-1，一般用于传输和钱相关的数据，对可靠性要求比较高的场景。

代码实现

// 设置 acks=-1
properties.put(ProducerConfig.ACKS_CONFIG, "all");
// 重试次数 retries，默认是 int 最大值，2147483647
properties.put(ProducerConfig.RETRIES_CONFIG, 3);

拓展：

image-20231227175355966

生产者将数据发送给 Leader，并且完成同步给 Follower，此时回复 ack 时，Leader 挂了，kafka 会挑一个 Follower 成为新的 Leader，因为生产者没有收到 ack，此时就会认为他的数据没有发送到 kafka，就会进行重试，导致新 Leader 重复接收了两份数据。

生产经验—数据去重

数据传递语义

image-20231227183602062

幂等性

幂等性原理

image-20231227183642104

如何使用幂等性

开启参数 enable.idempotence，默认为 true（默认开启）。

生产者事务

幂等性只能保证单分区单会话的不重复，一旦 kafka 挂掉重启，还是有可能产生重复数据。如果想完全去重，就必须使用事务。

Kafka 事务原理

image-20231227183743850

  • 幂等性：如果 kafka 挂掉重启，会重新生成一个 PID，所以可能会有重复。
  • 事务：kafka 根据全局唯一的 transactional.id 会划分到50个分区中的某一个分区，这些分区的信息是存储在一个特殊 Topic 里的，而 Topic 的底层就是硬盘，所以即使客户端挂掉了，重启后也能继续处理未完成的事务，因为有 transactional.id 存在。

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;

单个 Producer，使用事务保证消息的仅一次发送：

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 CustomProducerTransactions {
    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());

        // 设置事务 id（必须），事务 id 任意起名，要求全局唯一
        properties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "transaction_id_0");

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

        // 初始化事务
        kafkaProducer.initTransactions();

        // 开启事务
        kafkaProducer.beginTransaction();

        try {
            // 4. 调用 send 方法,发送消息
            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 {
            // 5. 关闭资源
            kafkaProducer.close();
        }
    }
}

生产经验—数据有序

image-20231228175454367

仅能保证单分区内有序，如果想保证全局有序，只能把所有分区的消息都拉到消费者端，进行一个全排序，再进行消费。

但需要等所有数据到齐了再进行排序，效率可能还不如单分区。

生产经验—数据乱序

一个 broker 可以有一个 broker 缓存队列，队列中存放的是还未收到 ack 的请求，最多能存放 5 个。

比如发送 Request1 后，对方没有应答，此时还可以发送 Request2、Request3、Request4、Request5，最多能发送 5 次请求。

假设在一个分区中，生产者发送了 Request1、Request2 请求都成功了，但 Request3 请求发送失败了，进行重试，但此时 Request4 请求发送成功了，然后 Request3 请求才发送成功，此时到达 kafka 的顺序就为 1 2 4 3，是乱序的。

  • kafka在1.x版本之前保证数据单分区有序，条件如下：
    • max.in.flight.requests.per.connection=1（不需要考虑是否开启幂等性）。
      • 也就是 broker 的缓存队列只允许有 1 个请求，这个请求收到 ack 后才能发送下一个。
  • kafka在1.x及以后版本保证数据单分区有序，条件如下：
    • 开启幂等性
      • max.in.flight.requests.per.connection 需要设置小于等于 5。
    • 未开启幂等性
      • max.in.flight.requests.per.connection 需要设置为 1（和kafka在1.x版本之前一样）。

原因说明：因为在 kafka1.x 以后，启用幂等后，kafka 服务端最多会缓存 producer 发来的最近 5 个 request 的元数据。

故无论如何，都可以保证最近 5 个 request 的数据都是有序的。

image-20231228180538249

  1. 比如先来的 Request1、Request2，服务端根据 SeqNumber 判断数据是否是单调递增的，如果符合则直接进行落盘；
  2. 但下一个请求是 Request4，正常应该是 Request3，所以 Request4 这个请求只能在内存中放着，不能进行落盘；
  3. 再下一个是 Request5，同样不能进行落盘；
  4. 直到 Request3 来了，然后对他们进行排序，然后再依次落盘 Request3、Request4、Request5。

笔记整理自b站尚硅谷视频教程：【尚硅谷】Kafka3.x教程（从入门到调优，深入全面）

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

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

相关文章

＜软考高项备考＞《论文专题 - 37 采购管理（2） 》

＜软考高项备考＞《论文专题 - 37 采购管理（2） 》

2 过程1-规划采购管理 2.1 问题 4W1H过程做什么记录项目采购决策、明确采购方法&#xff0c;及识别潜在卖方的过程作用&#xff1a;确定是否从项目外部获取货物和服务&#xff0c;如果是&#xff0c;则还要确定将在什么时间、以什么方式获取什么货物和服务为什么做为如何采购…
阅读更多...
移动端开发框架mui代码在安卓模拟器上运行（HbuilderX连接到模拟器）

移动端开发框架mui代码在安卓模拟器上运行（HbuilderX连接到模拟器）

开发工具 HBuilder X 3.8.12.20230817 注意&#xff1a;开发工具尽量用最新的或较新的。太旧的版本在开发调试过程中可能会出现莫名其妙的问题。 1、电脑下载安装安卓模拟器 我这里使用的是 夜神模拟器 &#xff0c;也可以选择其他安卓模拟器 夜神模拟器官网&#xff1a;夜神安…
阅读更多...
idea实现Java连接MySQL数据库

idea实现Java连接MySQL数据库

1.下载MySQL并安装 首先如果没有mysql的需要先下载MySQL&#xff0c;可以看这个教程&#xff1a; Mysql超详细安装配置教程(保姆级)_mysql安装及配置超详细教程-CSDN博客 2.下载mysql 的jdbc驱动 官网&#xff1a;MySQL :: Download Connector/J 解压并将驱动jar包导入id…
阅读更多...
ThinkPad T14s Gen3,ThinkPad X13 Gen3(21BS,21BQ,21BR,21BN)原装出厂Win11系统

ThinkPad T14s Gen3,ThinkPad X13 Gen3(21BS,21BQ,21BR,21BN)原装出厂Win11系统

lenovo联想ThinkPad系列T14s/X13 Gen3笔记本电脑原装Windows11预装OEM系统镜像 链接&#xff1a;https://pan.baidu.com/s/1yhRMIjlkFvt86aLioOoNOA?pwdfrsp 提取码&#xff1a;frsp 原厂系统自带所有驱动、出厂主题壁纸、系统属性专属联机支持标志、Office办公软件、联想…
阅读更多...
内网常规攻击路径

内网常规攻击路径

点击星标&#xff0c;即时接收最新推文 随着网络技术的发展&#xff0c;企业内部网络架构的变化&#xff0c;网络设备多样性的增加&#xff0c;面对内网攻击&#xff0c;防御体系逐渐阶梯化&#xff0c;通过不同维度的防御联动&#xff0c;将攻击拒之门外。对于突破网络边界后进…
阅读更多...
Rust学习笔记000 安装

Rust学习笔记000 安装

安装命令 curl --proto https --tlsv1.2 -sSf https://sh.rustup.rs | sh $ curl --proto https --tlsv1.2 -sSf https://sh.rustup.rs | sh info: downloading installerWelcome to Rust!This will download and install the official compiler for the Rust programming la…
阅读更多...
Goodbye2023, Hello 2024!

Goodbye2023, Hello 2024!

2023的所有比赛结束了&#xff0c;以后 xcpc 相关的比赛应该都和我没啥关系了&#xff0c;可能只打打蓝桥天梯了&#xff0c;等到明年的时候估计很多算法的东西也都忘记了吧&#xff0c;彻底退休了。打铁人不配叫退役&#xff0c;也不敢公开这篇文章&#xff0c;只敢在没人看的…
阅读更多...
【算法】数论---欧拉函数

【算法】数论---欧拉函数

什么是欧拉函数&#xff1f; 对于正整数n&#xff0c;欧拉函数是小于或等于n的正整数中与n互质的数的数目&#xff0c;记作φ(n) φ(1)1 当m,n互质时&#xff0c;φ(mn)φ(m)∗φ(n) 一、求一个正整数的欧拉函数---&#xff08;先对它分解质因数&#xff0c;然后套公式&#xf…
阅读更多...
2024新年贺词：从“第一次”到“每一次”，锐意进取，未来可期

2024新年贺词：从“第一次”到“每一次”，锐意进取，未来可期

从“第一次”到“每一次”&#xff1a;锐意进取&#xff0c;未来可期&#xff01; 2023年&#xff0c;对和数来讲&#xff0c;是很不平凡的一年。在这一年里&#xff0c;我们共同创造并见证了很多个“第一次”。2023年8月&#xff0c;SHANHAI国际版正式在泰国上线。第一次“出…
阅读更多...
LLM之RAG实战（九）| 高级RAG 03：多文档RAG体系结构

LLM之RAG实战（九）| 高级RAG 03：多文档RAG体系结构

在RAG&#xff08;检索和生成&#xff09;这样的框架内管理和处理多个文档有很大的挑战。关键不仅在于提取相关内容&#xff0c;还在于选择包含用户查询所寻求的信息的适当文档。基于用户查询对齐的多粒度特性&#xff0c;需要动态选择文档&#xff0c;本文将介绍结构化层次检索…
阅读更多...
python+opencv实现图片/短视频一键去水印

python+opencv实现图片/短视频一键去水印

目录 0 前言1 准备工作2 读取图片或视频3 添加回调获取鼠标绘制水印区域4 调用opencv函数5 绘制蒙版主循环6 去水印主循环总结 0 前言 在制作ppt个人文章或者分享图片过程中&#xff0c;经常会遇到一些带有水印的情况&#xff0c;不少人都希望能够去除这些水印&#xff0c;提高…
阅读更多...
分布式存储考点梳理 + 高频面试题

分布式存储考点梳理 + 高频面试题

欢迎来到分布式存储模环节&#xff0c;本文我将和你一起梳理面试中分布式系统的数据库的高频考点&#xff0c;做到温故知新。 面试中如何考察分布式存储 广义的分布式存储根据不同的应用领域&#xff0c;划分为以下的类别&#xff1a; 分布式协同系统 分布式文件系统 分布式…
阅读更多...
数据结构：单调栈

数据结构：单调栈

1.单调栈 单调栈是一种数据结构&#xff0c;其中存放的数据应该是有序的&#xff0c;所以单调栈也有单调递减栈和单调递增栈 单调递增栈&#xff1a;栈顶到栈底的元素大小是从小到大 单调递减栈&#xff1a;栈顶到栈底的元素大小是从大到小 单调栈主要就是用来求一个给定序列中…
阅读更多...
上海周边公路骑行路线分享，维乐带你抓住秋天的小尾巴

上海周边公路骑行路线分享，维乐带你抓住秋天的小尾巴

路线一&#xff1a;松江郊里骑行 　　 　　在魔都上海&#xff0c;藏着一条自然风景适宜&#xff0c;能眺望黄浦江的美丽骑行路线。导航到华长路杨家角就能到达起点&#xff0c;一路向西&#xff0c;这里路况非常好&#xff0c;只有一条小道&#xff0c;没有汽车的障碍&#xf…
阅读更多...
数组(定义,静态初始化,地址值,元素访问,索引,遍历,动态初始化,两种初始化的区别,练习)

数组(定义,静态初始化,地址值,元素访问,索引,遍历,动态初始化,两种初始化的区别,练习)

文章目录 1.数组概念&#xff1a; 2.数组的定义格式一&#xff1a;格式二&#xff1a;详解&#xff1a;注意点&#xff1a; 3.数组的静态初始化完整格式&#xff1a;格式详解&#xff1a;注意点&#xff1a;简化格式:练习1&#xff1a;练习2&#xff1a;练习3&#xff1a; 4.地…
阅读更多...
使用opencv+tesseract识别图片中的表格

使用opencv+tesseract识别图片中的表格

描述 在java环境中使用opencv和tesserac识别一个图片表格 环境&#xff1a;opencv和tesseract安装在linux环境下&#xff0c;docker将运行springboot服务 opencv和tesseract的安装和docker加载可参考之前的文章 过程 将图片进行预处理&#xff0c;过滤掉颜色等干扰元素 提…
阅读更多...
基于Ubuntu环境Git服务器搭建及使用

基于Ubuntu环境Git服务器搭建及使用

基于Ubuntu环境Git服务器搭建及使用 Chapter1 搭建本地git服务器及详细操作步骤1.搭建本地git服务器1.1 环境1.2 服务端配置1.3 创建git专属用户1.4 创建git仓库1.5 配置免密登录基础 2.客户端拉取推送代码2.1客户端创建ssh公钥 2.2 免密配置3.仓库使用&#xff08;拉取及推送代…
阅读更多...
记chrome的hackbar无法post php://input的问题

记chrome的hackbar无法post php://input的问题

尽管hackbar支持post请求体&#xff0c;但是当请求体里面没有等于号的时候&#xff0c;无法post出去&#xff0c;这样如果需要使用php://input绕过waf的时候就没法做。 在开发人员工具的网络里面可以看到不使用等于号的情况下没有荷载。 之后在这里看到了解决方法&#xff0c;…
阅读更多...
【javaSE】代理并不难

【javaSE】代理并不难

代理&#xff1a; 代理模式最主要的就是在不改变原来代码&#xff08;就是目标对象&#xff09;的情况下实现功能的增强 在学习AOP之前先了解代理&#xff0c;代理有两种&#xff1a;一种是动态代理&#xff0c;一类是静态代理。 静态代理 相当于是自己写了一个代理类&#…
阅读更多...
pandas将dataframe列中的list转换为多列

pandas将dataframe列中的list转换为多列

在应用机器学习的过程中&#xff0c;很大一部分工作都是在做数据的处理&#xff0c;一个非常常见的场景就是将一个list序列的特征数据拆成多个单独的特征数据。 比如数据集如下所示&#xff1a; data [[John, 25, Male,[99,100,98]],[Emily, 22, Female,[97,99,98]],[Michae…
阅读更多...
最新文章

Copyright @ 2022~2023