二、kafka生产与消费全流程

一、使用java代码生产、消费消息

1、生产者

package com.allwe.client.simple;import lombok.extern.slf4j.Slf4j;
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;/*** kafka生产者配置** @Author: AllWe* @Date: 2024/09/24/17:57*/
@Slf4j
public class HelloKafkaProducer {public static void main(String[] args) {// 设置属性Properties properties = new Properties();// 指定连接的kafka服务器地址，多台就用“,”隔开，如果某一台宕机生产者依然可以连接properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");// 设置key和value的序列化器，使java对象转换成二进制数组properties.put("key.serializer", StringSerializer.class);properties.put("value.serializer", StringSerializer.class);// new一个生产者producerKafkaProducer<String, String> producer = new KafkaProducer<>(properties);try {ProducerRecord<String, String> producerRecord;try {// 构建消息producerRecord = new ProducerRecord<>("topic_1", "student", "allwe");// 发送消息producer.send(producerRecord);System.out.println("消息发送成功");} catch (Exception e) {e.printStackTrace();}} finally {// 释放连接producer.close();}}
}

2、消费者

package com.allwe.client.simple;import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;import java.time.Duration;
import java.util.Collections;
import java.util.Properties;/*** kafka生产者配置** @Author: AllWe* @Date: 2024/09/24/17:57*/
@Slf4j
public class HelloKafkaConsumer {public static void main(String[] args) {// 设置属性Properties properties = new Properties();// 指定连接的kafka服务器地址，多台就用“,”隔开，如果某一台宕机生产者依然可以连接properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");// 设置key和value的序列化器，使java对象转换成二进制数组properties.put("key.deserializer", StringDeserializer.class);properties.put("value.deserializer", StringDeserializer.class);properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test");// new一个消费者consumerKafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);try {// 订阅哪些主题，可以多个，推荐订阅一个主题consumer.subscribe(Collections.singleton("topic_1"));// 死循环里面实现监听while (true) {// 每间隔1s，取一次消息，可能取到多条消息// 设置一秒的超时时间ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));for (ConsumerRecord<String, String> record : records) {System.out.println("key:" + record.key() + ",value:" + record.value());}}} finally {// 释放连接consumer.close();}}
}

3、踩坑

如果连接的不是本机的kafka，需要在目标机器的kafka配置文件中配置真实的ip地址，如果使用默认的配置或者配置为localhost:9092，kafka.clients会将目标机器的ip解析为127.0.0.1，导致连接不上kafka。

二、生产者

1、序列化器

在上面的demo中，由于消息的key和value都是String类型的，就可以使用kafka.client提供的String序列化器，如果想要发送其他自定义类型的对象，可以手动编写一个序列化器和反序列化器，实现Serializer接口，将对象和byte数组互相转换即可。

需要注意的是，生产者使用的自定义序列化器必须和消费者使用的反序列化器对应，否则无法正确解析消息。

那么什么情况下需要使用自定义序列化器呢？

        -- 需要兼容一些其他协议。

2、分区器

发送的消息被分配到哪个分区中？分区是如何选择的？假设上面的demo中，主题topic_1有4个分区，分别发送4次消息，处理分区的逻辑是怎样的？

这里需要先配置kafka在创建新的主题时，默认的分区数量，我这里配置为了4。

1）指定分区器

可以选择在创建生产者时，给生产者配置相关的分区器，指定具体分区算法。kafka.client提供了一些分区器，或者自己实现一个分区器。

// 设置分区规则
Properties properties = new Properties();
// 1、默认分区器
properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, DefaultPartitioner.class);
// 2、统一粘性分区器
properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, UniformStickyPartitioner.class);
// 3、自定义分区器
properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, MyPartitioner.class);

自定义分区器：

package com.allwe.client.partitioner;import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.utils.Utils;import java.util.List;
import java.util.Map;/*** 自定义分区器 - 以value值分区*/
public class MyPartitioner implements Partitioner {@Overridepublic int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {List<PartitionInfo> partitionInfoList = cluster.partitionsForTopic(topic);// 以value值的byte数组处理后再和分区数取模，决定放在哪个分区上return Utils.toPositive(Utils.murmur2(valueBytes)) % partitionInfoList.size();}@Overridepublic void close() {}@Overridepublic void configure(Map<String, ?> map) {}
}

2）指定分区

也可以选择在构建消息时指定分区，此时的分区优先级最高，不会被其他分区器影响。

# 创建消息时指定分区为 0
ProducerRecord<String, String> producerRecord = new ProducerRecord<>("topic_1", 0, "student", "allwe");

3、生产者发送消息的回调

package com.allwe.client.partitioner;import lombok.extern.slf4j.Slf4j;
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.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;import java.util.Properties;
import java.util.concurrent.Future;/*** kafka生产者配置 - 自定义分区器 & 发送消息回调** @Author: AllWe* @Date: 2024/09/24/17:57*/
@Slf4j
public class PartitionerProducer {public static void main(String[] args) {// 设置属性Properties properties = new Properties();// 指定连接的kafka服务器地址，多台就用“,”隔开，如果某一台宕机生产者依然可以连接properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");// 设置key和value的序列化器，使java对象转换成二进制数组properties.put("key.serializer", StringSerializer.class);properties.put("value.serializer", StringSerializer.class);// 设置自定义分区器properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, MyPartitioner.class);// new一个生产者producerKafkaProducer<String, String> producer = new KafkaProducer<>(properties);try {ProducerRecord<String, String> producerRecord;try {// 构建指定分区的消息，此时指定的分区不会变// producerRecord = new ProducerRecord<>("topic_1", 0, "student", "allwe");for (int i = 0; i < 10; i++) {// 构建消息producerRecord = new ProducerRecord<>("topic_2", "student", "allwe" + i);// 发送消息Future<RecordMetadata> future = producer.send(producerRecord);// 解析回调元数据RecordMetadata recordMetadata = future.get();System.out.println(i + ",offset:" + recordMetadata.offset() + ",partition:" + recordMetadata.partition());}} catch (Exception e) {e.printStackTrace();}} finally {// 释放连接producer.close();}}
}

打印结果：

4、异步解析生产者发送消息的回调

package com.allwe.client.callBack;import com.allwe.client.partitioner.MyPartitioner;
import lombok.extern.slf4j.Slf4j;
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;/*** kafka生产者配置 - 异步解析发送消息回调** @Author: AllWe* @Date: 2024/09/24/17:57*/
@Slf4j
public class AsynPartitionerProducer {public static void main(String[] args) {// 设置属性Properties properties = new Properties();// 指定连接的kafka服务器地址，多台就用“,”隔开，如果某一台宕机生产者依然可以连接properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");// 设置key和value的序列化器，使java对象转换成二进制数组properties.put("key.serializer", StringSerializer.class);properties.put("value.serializer", StringSerializer.class);// 设置自定义分区器properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, MyPartitioner.class);// new一个生产者producerKafkaProducer<String, String> producer = new KafkaProducer<>(properties);try {ProducerRecord<String, String> producerRecord;try {for (int i = 0; i < 10; i++) {// 构建消息producerRecord = new ProducerRecord<>("topic_3", "student", "allwe" + i);// 发送消息, 设置异步回调解析器producer.send(producerRecord, new CallBackImpl());}System.out.println("发送完成,topic_4");} catch (Exception e) {e.printStackTrace();}} finally {// 释放连接producer.close();}}
}

package com.allwe.client.callBack;import cn.hutool.core.util.ObjectUtil;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.RecordMetadata;/*** 异步发送消息回调解析器*/
public class CallBackImpl implements Callback {@Overridepublic void onCompletion(RecordMetadata recordMetadata, Exception e) {if (ObjectUtil.isNull(e)) {// 解析回调元数据System.out.println("offset:" + recordMetadata.offset() + ",partition:" + recordMetadata.partition());} else {e.printStackTrace();}}
}

5、生产者缓冲

1）为什么kafka在客户端发送消息的时候需要做一个缓冲？

① 减少IO的开销（单个 -> 批次），需要修改配置文件。

② 减少GC（核心）。

2）如何配置缓冲？

producer.properties配置文件中修改下面两个参数：

消息的大小：batch.size = 默认16384(16K) 

暂存的时间：linger.ms = 默认0ms

上面两个条件只要达到一个，就会发送消息，所以在默认配置下，生产一条消息就立即发送。

3）减少GC的原理

producer.properties配置文件的参数：

缓冲池大小：buffer.memory = 默认32M

kafka客户端使用了缓冲池，默认大小32M，当有一条新的消息进入缓冲池，达到了任何一个条件后就发送。发送后不用立即回收内存，而是初始化一下缓冲池即可，减少了GC的次数。

简单说就是利用池化技术减少了对象的创建 -> 减少内存分配次数 -> 减少了垃圾回收次数。

4）使用缓冲池的风险

当缓存的消息超出缓冲池的大小，kafka就会抛出OOM异常。

如果写入消息太快，但是上一次send方法没有执行完，就会导致上一次缓存的消息不能删除，这一次进来的消息又太多，最终写满了缓冲池，触发OOM异常。

解决办法就是适当调整buffer.memory参数和batch.size参数，增加缓冲池大小，缩小每一批次的大小。

三、Kafka Broker

消息从生产者发送出去后，就进入了broker中。在kafka broker中，每一个分区就是一个文件。

四、消费者

1、消费者群组

在消费的过程中，一般情况下使用群组消费，设置group_id_config。

核心：kafka群组消费的负载均衡建立在分区级别。

1）单个群组场景

一个分区只能由一个消费者消费。

在kafka执行过程中，支持动态添加或者减少消费者。

2）多个群组场景

群组之间的消费是互不干扰的，比如群组A的消费者和群组B的消费者可以同时消费同一个分区的消息。

2、Demo记录

写一个生产者，我为了测试顺畅写了一个无限循环的。只启动一次，输入参数即可实现批量发送消息。

package com.allwe.client.singleGroup;import com.allwe.client.partitioner.MyPartitioner;
import lombok.extern.slf4j.Slf4j;
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;
import java.util.Scanner;/*** kafka生产者配置 - 无限生产消息** @Author: AllWe* @Date: 2024/09/24/17:57*/
@Slf4j
public class Producer {public static void main(String[] args) {// 设置属性Properties properties = new Properties();properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");properties.put("key.serializer", StringSerializer.class);properties.put("value.serializer", StringSerializer.class);properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, MyPartitioner.class);// new一个生产者producerKafkaProducer<String, String> producer = new KafkaProducer<>(properties);Scanner scanner = new Scanner(System.in);;try {int count;while (true) {System.out.println("==================输入消息条数===================");String nextLine = scanner.nextLine();if ("exit".equals(nextLine)) {break;}count = Integer.parseInt(nextLine);ProducerRecord<String, String> producerRecord;try {for (int i = 0; i < count; i++) {// 构建消息producerRecord = new ProducerRecord<>("topic_5", "topic_5", "allwe" + i);producer.send(producerRecord);}} catch (Exception e) {e.printStackTrace();}System.out.println("发送完成,topic_5");}} catch (Exception e) {throw new RuntimeException(e);} finally {// 释放连接producer.close();scanner.close();}}
}
生产者控制台展示​​

写一个消费者base类，由于测试消费者需要启动很多类，我这里为了方便写了一个baseConsumer类，调用时new这个类的对象即可调用消费方法。

package com.allwe.client.singleGroup;import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;import java.time.Duration;
import java.util.Collections;
import java.util.Properties;/*** kafka 消费者配置** @Author: AllWe* @Date: 2024/09/24/17:57*/
@Slf4j
@Data
public class SingleGroupBaseConsumer {private String groupIdConfig;private String topicName;private KafkaConsumer<String, String> consumer;public SingleGroupBaseConsumer(String groupIdConfig, String topicName) {this.groupIdConfig = groupIdConfig;this.topicName = topicName;createConsumer();}private void createConsumer() {// 设置属性Properties properties = new Properties();properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");properties.put("key.deserializer", StringDeserializer.class);properties.put("value.deserializer", StringDeserializer.class);properties.put(ConsumerConfig.GROUP_ID_CONFIG, groupIdConfig);consumer = new KafkaConsumer<>(properties);}public void poll() {try {consumer.subscribe(Collections.singleton(topicName));while (true) {ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));int count = 0;for (ConsumerRecord<String, String> record : records) {count = 1;System.out.println("partition:" + record.partition() + ",key:" + record.key() + ",value:" + record.value());}if (count == 1) {// 消费到消息了就打印分隔线System.out.println("===============================");}}} finally {consumer.close();}}
}

 1）单个群组场景

群组id：allwe01

package com.allwe.client.singleGroup;import lombok.extern.slf4j.Slf4j;/*** kafka消费者启动器** @Author: AllWe* @Date: 2024/09/24/17:57*/
@Slf4j
public class SingleGroupConsumer_1 {public static void main(String[] args) {SingleGroupBaseConsumer singleGroupBaseConsumer = new SingleGroupBaseConsumer("allwe01", "topic_5");singleGroupBaseConsumer.poll();}
}
消费者控制台展示

我这里只放了一个消费者的消费记录，根据消费者控制台打印的数据，可以看到两条信息：

① 该消费者只能消费分区=1的消息。

② 消费者消费消息时，每次拿到的消息数量不确定。

2）多个群组场景

群组id：allwe02

package com.allwe.client.group;import com.allwe.client.singleGroup.SingleGroupBaseConsumer;
import lombok.extern.slf4j.Slf4j;/*** kafka消费者启动器** @Author: AllWe* @Date: 2024/09/24/17:57*/
@Slf4j
public class GroupConsumer_1 {public static void main(String[] args) {SingleGroupBaseConsumer singleGroupBaseConsumer = new SingleGroupBaseConsumer("allwe02", "topic_5");singleGroupBaseConsumer.poll();}
}
消费者控制台展示

可以看到，这里新加入了一个消费者群组，只有一个消费者，它就消费到了全部分区的消息。

3、ACK确认

消费者在成功消费消息后，会进行ACK确认。提交最后一次消费消息的偏移量，下一次消费就从上次提交的偏移量开始，如果一个新的消费者群组消费一个主题的消息，可以根据不同的配置来指定起始的偏移量。

// 从最早的消息开始消费
properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");// 从已提交的偏移量开始消费 - 默认配置
properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");

在kafka内部，有一个名字叫【__consumer_offsets】的主题，保存了消费者对各个主题的消费偏移量。消费者每一次发送的ACK确认，都会更新这个主题中的偏移量数据。

1）自动提交ACK的消费模式

默认的消费模式。

只要拿到了消息，就自动提交ACK确认。

但是有一个风险，就是虽然消费者成功取到了消息，但是在程序处理过程中出现了异常，同时提交了ACK确认，那么这条消息就永远不会被正确地处理。

所以有时候我们需要避免自动提交ACK确认，改成手动提交ACK确认。

2）手动提交ACK确认

取消自动提交

// 取消自动提交
properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
① 同步提交
// 同步提交ACK确认 - 提交不成功就一直重试，成功后才会继续往下执行
consumer.commitSync();

立刻进行ACK确认。但是容易造成阻塞，只有等待ACK确认成功后，才会继续执行程序。如果ACK确认不成功，就会一直重试。

② 异步提交
// 异步提交ACK确认
consumer.commitAsync();

异步提交不会阻塞应用程序，提交失败不会重试提交。

③ 组合使用demo
    public void poll() {try {consumer.subscribe(Collections.singleton(topicName));while (true) {ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));int count = 0;for (ConsumerRecord<String, String> record : records) {count = 1;System.out.println("partition:" + record.partition() + ",offset:" + record.offset() +",key:" + record.key() + ",value:" + record.value());}if (count == 1) {// 消费到消息了就打印分隔线System.out.println("===============================");}// 异步提交ACK确认consumer.commitAsync();}} finally {try {// 同步提交ACK确认 - 提交不成功就一直重试，成功后才会继续往下执行consumer.commitSync();} finally {consumer.close();}}}

3）手动批量提交ACK确认

如果消费者在某一时刻取到的消息数量太多，那么给每一条消息单独提交ACK确认太浪费资源，可以选择批量提交ACK确认。核心思想就是在程序中暂存偏移量，达到设定的阈值后就触发批量提交。

kafka.Consumer提供的异步提交ACK方法支持批量提交。

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

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

相关文章

C# 游戏引擎中的协程

C# 游戏引擎中的协程

前言 书接上回&#xff0c;我谈到了Unity中的协程的重要性&#xff0c;虽然协程不是游戏开发“必要的”&#xff0c;但是它可以在很多地方发挥优势。 为了在Godot找回熟悉的Unity协程开发手感&#xff0c;不得不自己做一个协程系统&#xff0c;幸运的是&#xff0c;有了Unity的…
阅读更多...
TI DSP TMS320F280025 Note15：串口SCI的使用

TI DSP TMS320F280025 Note15：串口SCI的使用

TMS320F280025 串口SCI的使用 ` 文章目录 TMS320F280025 串口SCI的使用框图分析串口特点可编程数据格式SCI端口中断非FIFO/FIFO模式下SCI中断的操作/配置UartDriver.cUartDriver.h串口时钟由PCLKCR7控制使能，默认位系统时钟4分频 串口接收与发送都可以触发中断 串口使用的引脚…
阅读更多...
JAVA并发编程高级——JDK 新增的原子操作类 LongAdder

JAVA并发编程高级——JDK 新增的原子操作类 LongAdder

LongAdder 简单介绍 前面讲过，AtomicLong通过CAS提供了非阻塞的原子性操作，相比使用阻塞算法的同步器来说它的性能已经很好了，但是JDK开发组并不满足于此。使用AtomicLong 时，在高并发下大量线程会同时去竞争更新同一个原子变量，但是由于同时只有一个线程的CAS操作会成功，…
阅读更多...
【C语言】指针篇 | 万字笔记

【C语言】指针篇 | 万字笔记

写在前面 在学习C语言过程&#xff0c;总有一个要点难点离不开&#xff0c;那就是大名鼎鼎的C语言指针&#xff0c;也是应为有指针的存在&#xff0c;使得C语言一直长盛不衰。因此不才把指针所学的所有功力都转换成这个笔记。希望对您有帮助&#x1f970;&#x1f970; 学习指…
阅读更多...
基于Springboot+Vue的饮食营养管理信息系统（含源码数据库）

基于Springboot+Vue的饮食营养管理信息系统（含源码数据库）

1.开发环境 开发系统:Windows10/11 架构模式:MVC/前后端分离 JDK版本: Java JDK1.8 开发工具:IDEA 数据库版本: mysql5.7或8.0 数据库可视化工具: navicat 服务器: SpringBoot自带 apache tomcat 主要技术: Java,Springboot,mybatis,mysql,vue 2.视频演示地址 3.功能 系统中…
阅读更多...
Hystrix学习

Hystrix学习

系列文章目录 JavaSE基础知识、数据类型学习万年历项目代码逻辑训练习题代码逻辑训练习题方法、数组学习图书管理系统项目面向对象编程&#xff1a;封装、继承、多态学习封装继承多态习题常用类、包装类、异常处理机制学习集合学习IO流、多线程学习仓库管理系统JavaSE项目员工…
阅读更多...
MAC M1 安装brew 配置环境变量，安装dart

MAC M1 安装brew 配置环境变量，安装dart

一. 下载 brew 1. 终端输入 /bin/bash -c "$(curl -fsSL https://raw.githubusercontent.com/Homebrew/install/HEAD/install.sh)" 2. 如遇到下载失败情况&#xff0c;需要VPN/代理 curl: (7) Failed to connect to raw.githubusercontent.com port 443 after 8 m…
阅读更多...
MongoDB简介

MongoDB简介

1、说到MongoDB就必须说下什么是NoSQL? NoSQL(NoSQL Not Only SQL)&#xff0c;意即反SQL运动&#xff0c;指的是非关系型的数据库&#xff0c;是一项全新的数据库革命性运动&#xff0c;早期就有人提出&#xff0c;发展至2009年趋势越发高涨。NoSQL的拥护者们提倡运用非关系…
阅读更多...
Java | Leetcode Java题解之第454题四数相加II

Java | Leetcode Java题解之第454题四数相加II

题目&#xff1a; 题解&#xff1a; class Solution {public int fourSumCount(int[] A, int[] B, int[] C, int[] D) {Map<Integer, Integer> countAB new HashMap<Integer, Integer>();for (int u : A) {for (int v : B) {countAB.put(u v, countAB.getOrDefa…
阅读更多...
被字节恶心到了

被字节恶心到了

字节 日常逛 xhs 看到一篇吐槽贴&#xff0c;表示被公司恶心到了&#xff1a; 这位网友表示&#xff0c;最近是公司举办了 Q2 和 H1 的优秀员工表彰&#xff0c;自己的 1&#xff08;直属领导&#xff09;评上了&#xff0c;但仔细一看&#xff0c;1 获奖的所有产出都是自己的&…
阅读更多...
实时数字人DH_live使用案例

实时数字人DH_live使用案例

参看： https://github.com/kleinlee/DH_live ubuntu 测试 apt install ffmpeg 下载安装： git clone https://github.com/kleinlee/DH_live.git cd DH_liveconda create -n dh_live python=3.12 conda activate dh_live pip install -r requirements.txt pip install torch …
阅读更多...
突发：Sam万字长文，OpenAI o1超越人类，o1模型训练原理、微调、能力来源-AI已死，大模型当立

突发：Sam万字长文，OpenAI o1超越人类，o1模型训练原理、微调、能力来源-AI已死，大模型当立

OpenAl o1大模型&#xff1a;原理、突破、前景及影响 北京时间2024年9月13日凌晨&#xff0c;OpenAI正式发布了新的人工智能模型o1&#xff08;o是orion猎户座&#xff0c;1代表从头再来&#xff0c;也意味着后续将出现更多序列&#xff09;&#xff0c;就是此前OpenAI一直在高…
阅读更多...
银河麒麟V10 SP1如何进入救援模式？

银河麒麟V10 SP1如何进入救援模式？

银河麒麟V10 SP1如何进入救援模式&#xff1f; 1、准备工作2、进入BIOS/UEFI进入救援模式注意事项 &#x1f496;The Begin&#x1f496;点点关注&#xff0c;收藏不迷路&#x1f496; 在使用银河麒麟高级服务器操作系统V10 SP1时&#xff0c;如果遇到系统无法正常启动或需要进…
阅读更多...
240 搜索二维矩阵 II

240 搜索二维矩阵 II

解题思路&#xff1a; \qquad 解这道题最重要的是如何利用从左到右、从上到下为升序的性质&#xff0c;快速找到目标元素。 \qquad 如果从左上角开始查找&#xff0c;如果当前matrix[i][[j] < target&#xff0c;可以向右、向下扩展元素都是升序&#xff0c;但选择哪个方向…
阅读更多...
.Net 6.0 监听Windows网络状态切换

.Net 6.0 监听Windows网络状态切换

上次发了一个文章获取windows网络状态&#xff0c;判断是否可以访问互联网。传送门&#xff1a;获取本机网络状态 这次我们监听网络状态切换&#xff0c;具体代码如下&#xff1a; public class WindowsNetworkHelper {private static Action<bool>? _NetworkStatusCh…
阅读更多...
初步认识产品经理

初步认识产品经理

产品经理 思考问题的维度 1️⃣为什么要抓住核心用户&#xff1f; 所有和产品有关系的群体就是用户&#xff0c;存在共性和差异了解用户的付费点&#xff0c;更好的优化产品是否使用&#xff1a;&#xff08;目标用户-已使用产品&#xff1a;种子用户-尝鲜&#xff1b;核心用…
阅读更多...
【在Linux世界中追寻伟大的One Piece】命名管道

【在Linux世界中追寻伟大的One Piece】命名管道

目录 1 -> 命名管道 1.1 -> 创建一个命名管道 1.2 -> 匿名管道与命名管道的区别 1.3 -> 命名管道的打开规则 1.4 -> 例子 1 -> 命名管道 管道应用的一个限制就是只能在具有共同祖先(具有亲缘关系)的进程间通信。如果我们想在不相关的进程之间交换数据&…
阅读更多...
C++多重继承

C++多重继承

C多重继承 一个类可以从多个类继承&#xff0c;只需在类的基类列表中&#xff08;即冒号后&#xff09;指定更多的基类&#xff0c;用逗号分隔即可。例如&#xff0c;如果程序有一个名为Output的特定类要在屏幕上打印&#xff0c;我们希望派生类Rectangle&#xff08;长方形&a…
阅读更多...
Netgear-WN604 downloadFile.php 信息泄露复现(CVE-2024-6646)

Netgear-WN604 downloadFile.php 信息泄露复现(CVE-2024-6646)

0x01 产品描述&#xff1a; NETGEAR WN604是一款功能强大的双频AC1200无线路由器,非常适合中大型家庭和企业使用。它支持最新的802.11ac无线标准,能提供高达1200Mbps的无线传输速度。路由器具备千兆有线网口和3个100Mbps有线网口,可满足有线和无线设备的接入需求。此外,它还内置…
阅读更多...
JavaWeb——Vue组件库Element（5/6）：案例：组件实现（概述、Form表单、Table表格、Pagination 分页、效果展示、完整代码）

JavaWeb——Vue组件库Element（5/6）：案例：组件实现（概述、Form表单、Table表格、Pagination 分页、效果展示、完整代码）

目录 概述 Form表单 Table表格 Pagination 分页 效果展示 完整代码 概述 在刚才制作出来的页面当中&#xff0c;上面项目的名称已制作好&#xff0c;左侧的菜单栏也已配置好。 接下来主要处理的是右侧主展示区域当中的组件编写。 在右侧的主展示区域&#xff0c;主要有…
阅读更多...
最新文章
推荐文章

Copyright @ 2022~2023