Spring-Kafka生产者源码分析

文章目录

    • 概要
    • 初始化
    • 消息发送
    • 小结

概要

本文主要概括Spring Kafka生产者发送消息的主流程

代码准备:
SpringBoot项目中maven填加以下依赖

<parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.3.12.RELEASE</version><relativePath/> <!-- lookup parent from repository -->
</parent>
<dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId>
</dependency>

消息发送使用KafkaTemplate

@Autowired
private KafkaTemplate<String, String> kafkaTemplate;@GetMapping("/test/send/{msg}")
public String sendMsg(@PathVariable String msg) {kafkaTemplate.send("alai_test", msg);return "success";
}

初始化

启动类KafkaAutoConfiguration
有两个地方需要关注

@Bean
@ConditionalOnMissingBean({KafkaTemplate.class})
public KafkaTemplate<?, ?> kafkaTemplate(ProducerFactory<Object, Object> kafkaProducerFactory, ProducerListener<Object, Object> kafkaProducerListener) {KafkaTemplate<Object, Object> kafkaTemplate = new KafkaTemplate(kafkaProducerFactory);if (this.messageConverter != null) {kafkaTemplate.setMessageConverter(this.messageConverter);}kafkaTemplate.setProducerListener(kafkaProducerListener);kafkaTemplate.setDefaultTopic(this.properties.getTemplate().getDefaultTopic());return kafkaTemplate;
}@Bean
@ConditionalOnMissingBean({ProducerFactory.class})
public ProducerFactory<?, ?> kafkaProducerFactory() {DefaultKafkaProducerFactory<?, ?> factory = new DefaultKafkaProducerFactory(this.properties.buildProducerProperties());String transactionIdPrefix = this.properties.getProducer().getTransactionIdPrefix();if (transactionIdPrefix != null) {factory.setTransactionIdPrefix(transactionIdPrefix);}return factory;
}

其中的ProducerFactory使用的是DefaultKafkaProducerFactory

在发送消息之前,Spring Kafka会先创建Producer,返回的是CloseSafeProducer实现类,在该类中有一个委托类Producer<K, V> delegate,真正的发送消息处理逻辑委托给KafkaProducerKafkaProducer实例构造如下,边幅原因,这里只展示需要说明的部分

KafkaProducer(Map<String, Object> configs,Serializer<K> keySerializer,Serializer<V> valueSerializer,ProducerMetadata metadata,KafkaClient kafkaClient,ProducerInterceptors<K, V> interceptors,Time time) {ProducerConfig config = new ProducerConfig(ProducerConfig.addSerializerToConfig(configs, keySerializer,valueSerializer));try {Map<String, Object> userProvidedConfigs = config.originals();this.producerConfig = config;this.time = time;String transactionalId = userProvidedConfigs.containsKey(ProducerConfig.TRANSACTIONAL_ID_CONFIG) ?(String) userProvidedConfigs.get(ProducerConfig.TRANSACTIONAL_ID_CONFIG) : null;this.clientId = config.getString(ProducerConfig.CLIENT_ID_CONFIG);LogContext logContext;if (transactionalId == null)logContext = new LogContext(String.format("[Producer clientId=%s] ", clientId));elselogContext = new LogContext(String.format("[Producer clientId=%s, transactionalId=%s] ", clientId, transactionalId));log = logContext.logger(KafkaProducer.class);log.trace("Starting the Kafka producer");Map<String, String> metricTags = Collections.singletonMap("client-id", clientId);MetricConfig metricConfig = new MetricConfig().samples(config.getInt(ProducerConfig.METRICS_NUM_SAMPLES_CONFIG)).timeWindow(config.getLong(ProducerConfig.METRICS_SAMPLE_WINDOW_MS_CONFIG), TimeUnit.MILLISECONDS).recordLevel(Sensor.RecordingLevel.forName(config.getString(ProducerConfig.METRICS_RECORDING_LEVEL_CONFIG))).tags(metricTags);List<MetricsReporter> reporters = config.getConfiguredInstances(ProducerConfig.METRIC_REPORTER_CLASSES_CONFIG,MetricsReporter.class,Collections.singletonMap(ProducerConfig.CLIENT_ID_CONFIG, clientId));reporters.add(new JmxReporter(JMX_PREFIX));this.metrics = new Metrics(metricConfig, reporters, time);this.partitioner = config.getConfiguredInstance(ProducerConfig.PARTITIONER_CLASS_CONFIG, Partitioner.class);long retryBackoffMs = config.getLong(ProducerConfig.RETRY_BACKOFF_MS_CONFIG);if (keySerializer == null) {this.keySerializer = config.getConfiguredInstance(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,Serializer.class);this.keySerializer.configure(config.originals(), true);} else {config.ignore(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG);this.keySerializer = keySerializer;}if (valueSerializer == null) {this.valueSerializer = config.getConfiguredInstance(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,Serializer.class);this.valueSerializer.configure(config.originals(), false);} else {config.ignore(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG);this.valueSerializer = valueSerializer;}// load interceptors and make sure they get clientIduserProvidedConfigs.put(ProducerConfig.CLIENT_ID_CONFIG, clientId);ProducerConfig configWithClientId = new ProducerConfig(userProvidedConfigs, false);List<ProducerInterceptor<K, V>> interceptorList = (List) configWithClientId.getConfiguredInstances(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, ProducerInterceptor.class);// 生产者拦截器       if (interceptors != null)this.interceptors = interceptors;elsethis.interceptors = new ProducerInterceptors<>(interceptorList);ClusterResourceListeners clusterResourceListeners = configureClusterResourceListeners(keySerializer,valueSerializer, interceptorList, reporters);// 生产者往服务端发送消息的时候,规定一条消息最大多大?// 如果你超过了这个规定消息的大小,你的消息就不能发送过去。// 默认是1M,这个值偏小,在生产环境中,我们需要修改这个值。// 经验值是10M。但是大家也可以根据自己公司的情况来。       this.maxRequestSize = config.getInt(ProducerConfig.MAX_REQUEST_SIZE_CONFIG);//指的是缓存大小//默认值是32M,这个值一般是够用,如果有特殊情况的时候,我们可以去修改这个值。this.totalMemorySize = config.getLong(ProducerConfig.BUFFER_MEMORY_CONFIG);// kafka是支持压缩数据的,可以设置压缩格式,默认是不压缩,支持gzip、snappy、lz4// 一次发送出去的消息就更多。生产者这儿会消耗更多的cpu.this.compressionType = CompressionType.forName(config.getString(ProducerConfig.COMPRESSION_TYPE_CONFIG));// 配置控制了KafkaProducer.send()并将KafkaProducer.partitionsFor()被阻塞多长时间,由于缓冲区已满或元数据不可用,这些方法可// 能会被阻塞止this.maxBlockTimeMs = config.getLong(ProducerConfig.MAX_BLOCK_MS_CONFIG);int deliveryTimeoutMs = configureDeliveryTimeout(config, log);this.apiVersions = new ApiVersions();this.transactionManager = configureTransactionState(config, logContext);// 创建核心组件:记录累加器this.accumulator = new RecordAccumulator(logContext,config.getInt(ProducerConfig.BATCH_SIZE_CONFIG),this.compressionType,lingerMs(config),retryBackoffMs,deliveryTimeoutMs,metrics,PRODUCER_METRIC_GROUP_NAME,time,apiVersions,transactionManager,new BufferPool(this.totalMemorySize, config.getInt(ProducerConfig.BATCH_SIZE_CONFIG), metrics, time, PRODUCER_METRIC_GROUP_NAME));List<InetSocketAddress> addresses = ClientUtils.parseAndValidateAddresses(config.getList(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG),config.getString(ProducerConfig.CLIENT_DNS_LOOKUP_CONFIG));if (metadata != null) {this.metadata = metadata;} else {// 生产者每隔一段时间都要去更新一下集群的元数据,默认5分钟this.metadata = new ProducerMetadata(retryBackoffMs,config.getLong(ProducerConfig.METADATA_MAX_AGE_CONFIG),config.getLong(ProducerConfig.METADATA_MAX_IDLE_CONFIG),logContext,clusterResourceListeners,Time.SYSTEM);this.metadata.bootstrap(addresses);}this.errors = this.metrics.sensor("errors");// 真正执行消息发送的逻辑this.sender = newSender(logContext, kafkaClient, this.metadata);String ioThreadName = NETWORK_THREAD_PREFIX + " | " + clientId;this.ioThread = new KafkaThread(ioThreadName, this.sender, true);// 开启新的线程this.ioThread.start();config.logUnused();AppInfoParser.registerAppInfo(JMX_PREFIX, clientId, metrics, time.milliseconds());log.debug("Kafka producer started");} catch (Throwable t) {// call close methods if internal objects are already constructed this is to prevent resource leak. see KAFKA-2121close(Duration.ofMillis(0), true);// now propagate the exceptionthrow new KafkaException("Failed to construct kafka producer", t);}}

创建Sender时的方法如下:

// visible for testing
Sender newSender(LogContext logContext, KafkaClient kafkaClient, ProducerMetadata metadata) {// 使用幂等性,需要将 enable.idempotence 配置项设置为true。并且它对单个分区的发送,一次性最多发送5条int maxInflightRequests = producerConfig.getInt(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION);// 控制客户端等待请求响应的最长时间。如果在超时过去之前未收到响应,客户端将// 在必要时重新发送请求,或者如果重试耗尽,请求失败int requestTimeoutMs = producerConfig.getInt(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG);ChannelBuilder channelBuilder = ClientUtils.createChannelBuilder(producerConfig, time, logContext);ProducerMetrics metricsRegistry = new ProducerMetrics(this.metrics);Sensor throttleTimeSensor = Sender.throttleTimeSensor(metricsRegistry.senderMetrics);// 初始化了一个重要的管理网路的组件
// connections.max.idle.ms: 默认值是9分钟, 一个网络连接最多空闲多久,超过这个空闲时间,就关闭这个网络连接。
// max.in.flight.requests.per.connection:默认是5, producer向broker发送数据的时候,其实是有多个网络连接。每个网络连接可以忍受 producer端发送给broker 消息然后消息没有响应的个数KafkaClient client = kafkaClient != null ? kafkaClient : new NetworkClient(new Selector(producerConfig.getLong(ProducerConfig.CONNECTIONS_MAX_IDLE_MS_CONFIG),this.metrics, time, "producer", channelBuilder, logContext),metadata,clientId,maxInflightRequests,producerConfig.getLong(ProducerConfig.RECONNECT_BACKOFF_MS_CONFIG),producerConfig.getLong(ProducerConfig.RECONNECT_BACKOFF_MAX_MS_CONFIG),producerConfig.getInt(ProducerConfig.SEND_BUFFER_CONFIG),producerConfig.getInt(ProducerConfig.RECEIVE_BUFFER_CONFIG),requestTimeoutMs,producerConfig.getLong(ProducerConfig.SOCKET_CONNECTION_SETUP_TIMEOUT_MS_CONFIG),producerConfig.getLong(ProducerConfig.SOCKET_CONNECTION_SETUP_TIMEOUT_MAX_MS_CONFIG),time,true,apiVersions,throttleTimeSensor,logContext);short acks = Short.parseShort(producerConfig.getString(ProducerConfig.ACKS_CONFIG));return new Sender(logContext,client,metadata,this.accumulator,maxInflightRequests == 1,producerConfig.getInt(ProducerConfig.MAX_REQUEST_SIZE_CONFIG),acks,producerConfig.getInt(ProducerConfig.RETRIES_CONFIG), // 重试次数metricsRegistry.senderMetrics,time,requestTimeoutMs,producerConfig.getLong(ProducerConfig.RETRY_BACKOFF_MS_CONFIG),this.transactionManager,apiVersions);
}

在创建RecordAccumulator时,其内部会维护一个ConcurrentMap<TopicPartition, Deque<ProducerBatch>> batches ,该Map的key是TopicPartition,这个类重写了equals方法,相同的topic,相同的分区,在batches中属于相同的key,就会被放入到队列Deque中。

消息发送

Spring Kafka对消息的发送,最后也是直接委托给了org.apache.kafka.clients.producer.KafkaProducer#doSend方法,下面以这个方法作为入口进行分析

@Override
public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback) {// intercept the record, which can be potentially modified; this method does not throw exceptionsProducerRecord<K, V> interceptedRecord = this.interceptors.onSend(record);return doSend(interceptedRecord, callback);
}

拦截器

onSend 方法是遍历拦截器 onSend 方法,拦截器的目的是将数据处理加工, kafka 本身并没有给出默认的拦截器的实现。如果需要使用拦截器功能,必须自己实现 ProducerInterceptor 接口

public ProducerRecord<K, V> onSend(ProducerRecord<K, V> record) {ProducerRecord<K, V> interceptRecord = record;for (ProducerInterceptor<K, V> interceptor : this.interceptors) {try {interceptRecord = interceptor.onSend(interceptRecord);} catch (Exception e) {// 其中一个拦截器出现处理异常时不回抛出异常,只会打印日志// do not propagate interceptor exception, log and continue calling other interceptors// be careful not to throw exception from hereif (record != null)log.warn("Error executing interceptor onSend callback for topic: {}, partition: {}", record.topic(), record.partition(), e);elselog.warn("Error executing interceptor onSend callback", e);}}return interceptRecord;
}

ProducerInterceptor的3个方法:

  • onSend: Producer确保在消息被序列化以计算分区前调用该方法。用户可以在该方法中对消息做任何操作,但最好保证不要修改消息所属的topic和分区,否则会影响目标分区的计算
  • onAcknowledgement: 该方法会在消息被应答之前或消息发送失败时调用,并且通常都是在producer回调逻辑触发之前。onAcknowledgement运行在producer的IO线程中,因此不要在该方法中放入很重的逻辑,否则会拖慢producer的消息发送效率
  • close: 关闭interceptor,主要用于执行一些资源清理工作

消息发送主流程

/*** Implementation of asynchronously send a record to a topic.*/
private Future<RecordMetadata> doSend(ProducerRecord<K, V> record, Callback callback) {TopicPartition tp = null;try {throwIfProducerClosed();// first make sure the metadata for the topic is availablelong nowMs = time.milliseconds();ClusterAndWaitTime clusterAndWaitTime;try {// 首先确保该topic的元数据可用clusterAndWaitTime = waitOnMetadata(record.topic(), record.partition(), nowMs, maxBlockTimeMs);} catch (KafkaException e) {if (metadata.isClosed())throw new KafkaException("Producer closed while send in progress", e);throw e;}nowMs += clusterAndWaitTime.waitedOnMetadataMs;long remainingWaitMs = Math.max(0, maxBlockTimeMs - clusterAndWaitTime.waitedOnMetadataMs);Cluster cluster = clusterAndWaitTime.cluster;// 序列化 record 的 key 和 valuebyte[] serializedKey;try {serializedKey = keySerializer.serialize(record.topic(), record.headers(), record.key());} catch (ClassCastException cce) {throw new SerializationException("Can't convert key of class " + record.key().getClass().getName() +" to class " + producerConfig.getClass(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG).getName() +" specified in key.serializer", cce);}byte[] serializedValue;try {serializedValue = valueSerializer.serialize(record.topic(), record.headers(), record.value());} catch (ClassCastException cce) {throw new SerializationException("Can't convert value of class " + record.value().getClass().getName() +" to class " + producerConfig.getClass(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG).getName() +" specified in value.serializer", cce);}// 获取该 record 要发送到的 partitionint partition = partition(record, serializedKey, serializedValue, cluster);tp = new TopicPartition(record.topic(), partition);setReadOnly(record.headers());Header[] headers = record.headers().toArray();int serializedSize = AbstractRecords.estimateSizeInBytesUpperBound(apiVersions.maxUsableProduceMagic(),compressionType, serializedKey, serializedValue, headers);ensureValidRecordSize(serializedSize);long timestamp = record.timestamp() == null ? nowMs : record.timestamp();if (log.isTraceEnabled()) {log.trace("Attempting to append record {} with callback {} to topic {} partition {}", record, callback, record.topic(), partition);}// producer callback will make sure to call both 'callback' and interceptor callbackCallback interceptCallback = new InterceptorCallback<>(callback, this.interceptors, tp);// 向 accumulator 中追加 record 数据,数据会先进行缓存RecordAccumulator.RecordAppendResult result = accumulator.append(tp, timestamp, serializedKey,serializedValue, headers, interceptCallback, remainingWaitMs, true, nowMs);if (result.abortForNewBatch) {int prevPartition = partition;partitioner.onNewBatch(record.topic(), cluster, prevPartition);partition = partition(record, serializedKey, serializedValue, cluster);tp = new TopicPartition(record.topic(), partition);if (log.isTraceEnabled()) {log.trace("Retrying append due to new batch creation for topic {} partition {}. The old partition was {}", record.topic(), partition, prevPartition);}// producer callback will make sure to call both 'callback' and interceptor callbackinterceptCallback = new InterceptorCallback<>(callback, this.interceptors, tp);result = accumulator.append(tp, timestamp, serializedKey,serializedValue, headers, interceptCallback, remainingWaitMs, false, nowMs);}if (transactionManager != null) {transactionManager.maybeAddPartition(tp);}// 如果追加完数据后,对应的 RecordBatch 已经达到了 batch.size 的大小(或者batch 的剩余空间不足以添加下一条 Record),则唤醒 sender 线程发送数据。if (result.batchIsFull || result.newBatchCreated) {log.trace("Waking up the sender since topic {} partition {} is either full or getting a new batch", record.topic(), partition);this.sender.wakeup();}return result.future;} catch (ApiException e) {...}...
}
  • Producer 通过 waitOnMetadata() 方法来获取对应 topic 的 metadata 信息,需要先该topic 是可用的

  • Producer 端对 recordkeyvalue 值进行序列化操作,在 Consumer 端再进行相应的反序列化

  • 获取partition值,具体分为下面三种情况:
    1 指明 partition 的情况下,直接将指明的值直接作为 partiton 值
    2 没有指明 partition 值但有 key 的情况下,将 key 的 hash 值与 topic 的 partition 数进行取余得到 partition 值
    3 既没有 partition 值又没有 key 值的情况下,第一次调用时随机生成一个整数(后面每次调用在这个整数上自增),将这个值与 topic 可用的 partition 总数取余得到partition 值,也就是常说的 round-robin 算法
    4 Producer 默认使用的 partitioner 是org.apache.kafka.clients.producer.internals.DefaultPartitioner

  • accumulator 写数据,先将 record 写入到 buffer 中,当达到一个 batch.size 的大小时,再唤起 sender线程去发送 RecordBatch,这里仔细分析一下Producer是如何向buffer写入数据的
    1.获取该 topic-partition 对应的 queue,没有的话会创建一个空的 queue
    2.向 queue 中追加数据,先获取 queue 中最新加入的那个 RecordBatch,如果不存在或者存在但剩余空余不足以添加本条 record 则返回 null,成功写入的话直接返回结果,写入成功
    3.创建一个新的 RecordBatch,初始化内存大小根据 max(batch.size,Records.LOG_OVERHEAD + Record.recordSize(key, value)) 来确定(防止单条record 过大的情况)
    4.向新建的 RecordBatch 写入 record,并将 RecordBatch 添加到 queue 中,返回结果,写入成功

  • 发送 RecordBatch,当 record 写入成功后,如果发现 RecordBatch 已满足发送的条件(通常是 queue 中有多个 batch,那么最先添加的那些 batch 肯定是可以发送了),那么就会唤醒sender 线程,发送 RecordBatch 。sender 线程对 RecordBatch 的处理是在 run() 方法中进行的,该方法具体实现如下:
    1.获取那些已经可以发送的 RecordBatch 对应的 nodes
    2.如果与node 没有连接(如果可以连接,同时初始化该连接),就证明该 node 暂时不能发送数据,暂时移除该 node
    3.返回该 node 对应的所有可以发送的 RecordBatch 组成的 batches(key 是
    node.id),并将 RecordBatch 从对应的 queue 中移除
    4.将由于元数据不可用而导致发送超时的 RecordBatch 移除
    5.发送 RecordBatch

在这里插入图片描述

小结

在这里插入图片描述

由上图可以看出:KafkaProducer有两个基本线程:

主线程:

  1. 负责消息创建,拦截器,序列化器,分区器等操作,并将消息追加到消息收集器RecordAccumulator中;
  2. RecordAccumulator为每个分区都维护了一个Deque` 类型的双端队列。
  3. ProducerBatch可以理解为是ProducerRecord` 的集合,批量发送有利于提升吞吐量,降低网络影响;
  4. 由于生产者客户端使用 java.io.ByteBuffer 在发送消息之前进行消息保存,并维护了一个 BufferPool 实现 ByteBuffer 的复用;该缓存池只针对特定大小( batch.size指定)的 ByteBuffer进行管理,对于消息过大的缓存,不能做到重复利用。
  5. 每次追加一条ProducerRecord消息,会寻找/新建对应的双端队列,从其尾部获取一个ProducerBatch,判断当前消息的大小是否可以写入该批次中。若可以写入则写入;若不可以写入,则新建一个ProducerBatch,判断该消息大小是否超过客户端参数配置 batch.size 的值,不超过,则以 batch.size建立新的ProducerBatch,这样方便进行缓存重复利用;若超过,则以计算的消息大小建立对应的 ProducerBatch ,缺点就是该内存不能被复用了。

Sender线程:

  1. 该线程从消息收集器获取缓存的消息,将其处理为 <Node, List<ProducerBatch> 的形式, Node 表示集群的broker节点。
  2. 进一步将<Node, List<ProducerBatch>转化为<Node, Request>形式,此时才可以向服务端发送数据。
  3. 在发送之前,Sender线程将消息以 Map<NodeId, Deque<Request>> 的形式保存到InFlightRequests 中进行缓存,可以通过其获取 leastLoadedNode ,即当前Node中负载压力最小的一个,以实现消息的尽快发出。

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

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

相关文章

Flink提交jar出现错误RestHandlerException: No jobs included in application.

今天打包一个flink的maven工程为jar&#xff0c;通过flink webUI提交&#xff0c;发现居然报错。 如上图所示&#xff0c;提示错误为&#xff1a; Server Response Message: org.apache.flink.runtime.rest.handler.RestHandlerException: No jobs included in application. …

Java 线程池概念总结(thread pool)

一、池化思想的应用 池化思想是一种常见软件设计和优化技术。以下是几个常见池化思想应用场景&#xff1a; 线程池&#xff1a;线程池是池化思想一个典型应用。通过预先创建一组线程并将它们置于就绪状态&#xff0c;以复用线程减少线程创建和销毁的开销&#xff0c;并提高系统…

Nginx特性应用及载装

Nginx是一款轻量级的Web 服务器/反向代理服务器及电子邮件(IMAP/POP3)代理服务器。其特点是占有内存少&#xff0c;并发能力强&#xff0c;事实上nginx的并发能力在同类型的网页服务器中表现较好&#xff0c;中国大陆使用nginx的网站有&#xff1a;网易、腾讯、阿里等。 …

MyBatis-Plus —— 初窥门径

前言 在前面的文章中荔枝梳理了MyBatis及相关的操作&#xff0c;作为MyBatis的增强工具&#xff0c;MyBatis-Plus无需再在xml中写sql语句&#xff0c;在这篇文章中荔枝将梳理MyBatis-Plus的基础知识并基于SpringBoot梳理MyBatis-Plus给出的两个接口&#xff1a;BaseMapper和ISe…

对象模型和this指针(个人学习笔记黑马学习)

1、成员变量和成员函数 #include <iostream> using namespace std; #include <string>//成员变量和成员函数分开存储class Person {int m_A;//非静态成员变量 属于类的对象上的static int m_B;//静态成员变量 不属于类的对象上void func() {} //非静态成员函数 不…

供应链 | 顶会CIKM论文精读:面向大规模三维装箱问题的数据驱动树形搜索算法

论文解读&#xff1a;丁建辉&#xff0c;李明哲&#xff0c;赵艳蓉&#xff0c;孙楚天 编者按 本次解读的文章发表于CCF-B类会议30th ACM International Conference on Information and Knowledge Management。摘要总结如下&#xff1a; 3维装箱问题&#xff08;3D-BPP&#…

Redis的缓存穿透,缓存击穿,缓存雪崩

1. 缓存穿透 什么是缓存穿透&#xff1f; 缓存穿透说简单点就是大量请求的 key 是不合理的&#xff0c;根本不存在于缓存中&#xff0c;也不存在于数据库中 。这就导致这些请求直接到了数据库上&#xff0c;根本没有经过缓存这一层&#xff0c;对数据库造成了巨大的压力&…

雪花算法生成id分析与实践

目录 1 什么是雪花算法&#xff1f; 结构 优点 缺点 2 在java中使用 使用注意&#xff1a; 测试代码 效果 1 什么是雪花算法&#xff1f; witter的雪花算法&#xff08;Snowflake Algorithm&#xff09;。雪花ID是一种分布式唯一ID生成算法&#xff0c;旨在解决分布式…

数学建模-点评笔记 9月3日

1.摘要&#xff1a;关键方法和结论&#xff08;精炼的语言&#xff09;要说明&#xff0c;方法的合理性和意义也可以说明。 评委先通过摘要筛选&#xff08;第一轮&#xff09; 2.时间序列找异常值除了3西格玛还有针对时间序列更合适寻找的方法 3.模型的优缺点要写的详细一点…

世微AP9234 升压型DC/DC LED恒流驱动

描述 AP9234是一款由基准电压源、振荡电路、误差放大电路、相位补偿电路、电流限制电路等构成的CMOS升压型DC/DC LED驱动。由于内置了低导通电阻的增强型N沟道功率MOSFET&#xff0c;因此适用于需要高效率、高输出电流的应用电路。另外&#xff0c;可通过在VSENSE端子连接电流…

754. 到达终点数字

754. 到达终点数字 原题链接&#xff1a;完成情况&#xff1a;解题思路&#xff1a;参考代码&#xff1a; 原题链接&#xff1a; 754. 到达终点数字 https://leetcode.cn/problems/reach-a-number/description/ 完成情况&#xff1a; 解题思路&#xff1a; 牛顿莱布尼茨梯…

SWAT-MODFLOW地表水与地下水耦合

耦合模型被应用到很多科学和工程领域来改善模型的性能、效率和结果&#xff0c;SWAT作为一个地表水模型可以较好的模拟主要的水文过程&#xff0c;包括地表径流、降水、蒸发、风速、温度、渗流、侧向径流等&#xff0c;但是对于地下水部分的模拟相对粗糙&#xff0c;考虑到SWAT…

ConsoleApplication815项目(直接加载+VEH Hook Load)

上线图 ConsoleApplication815.cpp #include <iostream> #include<Windows.h> #include "detours.h" #include "detver.h" #pragma comment(lib,"detours.lib")#pragma warning(disable:4996)LPVOID Beacon_address; SIZE_T Beacon…

敏捷开发、V模型开发、瀑布模型

在软件开发领域&#xff0c;敏捷开发和V模型开发是两种主要的开发方法。它们之间的差异主要体现在开发过程的结构和组织方式上。在以下讨论中&#xff0c;我们将深入探讨这两种方法的特点和差异。 敏捷开发 敏捷开发是一种迭代和增量的软件开发方法&#xff0c;它强调灵活性和…

rz命令无法正常使用?

使用rz命令上传文件时出现如下问题&#xff1a; 这里用的是mobaxterm终端 改用xshell,secureCRT即可正常使用&#xff1a;

vscode调教配置:快捷修复和格式化代码

配置vscode快捷键&#xff0c;让你像使用idea一样使用vscode&#xff0c;我们最常用的两个功能就是格式化代码和快捷修复&#xff0c;所以这里修改一下快捷修复和格式化代码的快捷键。 在设置中&#xff0c;找到快捷键配置&#xff1a; 然后搜索&#xff1a;快捷修复 在快捷键…

Mqtt学习笔记--交叉编译移植(1)

简述 Mqtt目前在物联网行业的应用比较多&#xff0c;mqtt属于应用层的一个中间件&#xff0c;这个中间件实现消息的订阅发布机制。网上介绍Mqtt的实现原来的比较多&#xff0c;这里不细介绍。 其实在我们之前的产品中&#xff0c;自己也开发的有类似的中间件&#xff0c;除了具…

ORB-SLAM2算法12之单目初始化Initializer

文章目录 0 引言1 单目初始化Initializer1.1 构造函数1.2 成员函数1.2.1 Initialize1.2.2 FindHomography1.2.3 FindFundamental1.2.4 ReconstructH1.2.5 ReconstructF 2 总结 0 引言 ORB-SLAM2算法7详细了解了System主类和多线程、ORB-SLAM2学习笔记8详细了解了图像特征点提取…

每日一题 1921. 消灭怪物的最大数量

难度&#xff1a;中等 思路&#xff1a; 已知速度和距离&#xff0c;可求时间必定先消灭时间最短的怪物求得时间数组排序&#xff0c;只要在第 i 秒时&#xff0c;time[i] > i &#xff0c;那么就可以消灭第 i 个怪物 代码&#xff1a; class Solution:def eliminateMax…

QT(9.3)定时器,绘制事件

作业&#xff1a; 自定义一个闹钟 pro文件&#xff1a; QT core gui texttospeechgreaterThan(QT_MAJOR_VERSION, 4): QT widgetsCONFIG c11# The following define makes your compiler emit warnings if you use # any Qt feature that has been marked deprecat…