聊聊KafkaListener的实现机制
序
本文只要研究一下KafkaListener的实现机制
KafkaListener
org/springframework/kafka/annotation/KafkaListener.java
@Target({ ElementType.TYPE, ElementType.METHOD, ElementType.ANNOTATION_TYPE })
@Retention(RetentionPolicy.RUNTIME)
@MessageMapping
@Documented
@Repeatable(KafkaListeners.class)
public @interface KafkaListener {String id() default "";String containerFactory() default "";String[] topics() default {};String topicPattern() default "";TopicPartition[] topicPartitions() default {};String containerGroup() default "";String errorHandler() default "";String groupId() default "";boolean idIsGroup() default true;String clientIdPrefix() default "";String beanRef() default "__listener";String concurrency() default "";String autoStartup() default "";String[] properties() default {};
}
KafkaListener注解定义了id、topics、groupId等属性
KafkaListenerAnnotationBeanPostProcessor
org/springframework/kafka/annotation/KafkaListenerAnnotationBeanPostProcessor.java
public class KafkaListenerAnnotationBeanPostProcessor<K, V>implements BeanPostProcessor, Ordered, BeanFactoryAware, SmartInitializingSingleton {private final KafkaListenerEndpointRegistrar registrar = new KafkaListenerEndpointRegistrar(); @Overridepublic int getOrder() {return LOWEST_PRECEDENCE;}@Overridepublic void setBeanFactory(BeanFactory beanFactory) {this.beanFactory = beanFactory;if (beanFactory instanceof ConfigurableListableBeanFactory) {this.resolver = ((ConfigurableListableBeanFactory) beanFactory).getBeanExpressionResolver();this.expressionContext = new BeanExpressionContext((ConfigurableListableBeanFactory) beanFactory,this.listenerScope);}}@Overridepublic void afterSingletonsInstantiated() {this.registrar.setBeanFactory(this.beanFactory);if (this.beanFactory instanceof ListableBeanFactory) {Map<String, KafkaListenerConfigurer> instances =((ListableBeanFactory) this.beanFactory).getBeansOfType(KafkaListenerConfigurer.class);for (KafkaListenerConfigurer configurer : instances.values()) {configurer.configureKafkaListeners(this.registrar);}}if (this.registrar.getEndpointRegistry() == null) {if (this.endpointRegistry == null) {Assert.state(this.beanFactory != null,"BeanFactory must be set to find endpoint registry by bean name");this.endpointRegistry = this.beanFactory.getBean(KafkaListenerConfigUtils.KAFKA_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME,KafkaListenerEndpointRegistry.class);}this.registrar.setEndpointRegistry(this.endpointRegistry);}if (this.defaultContainerFactoryBeanName != null) {this.registrar.setContainerFactoryBeanName(this.defaultContainerFactoryBeanName);}// Set the custom handler method factory once resolved by the configurerMessageHandlerMethodFactory handlerMethodFactory = this.registrar.getMessageHandlerMethodFactory();if (handlerMethodFactory != null) {this.messageHandlerMethodFactory.setMessageHandlerMethodFactory(handlerMethodFactory);}else {addFormatters(this.messageHandlerMethodFactory.defaultFormattingConversionService);}// Actually register all listenersthis.registrar.afterPropertiesSet();}@Overridepublic Object postProcessAfterInitialization(final Object bean, final String beanName) throws BeansException {if (!this.nonAnnotatedClasses.contains(bean.getClass())) {Class<?> targetClass = AopUtils.getTargetClass(bean);Collection<KafkaListener> classLevelListeners = findListenerAnnotations(targetClass);final boolean hasClassLevelListeners = classLevelListeners.size() > 0;final List<Method> multiMethods = new ArrayList<>();Map<Method, Set<KafkaListener>> annotatedMethods = MethodIntrospector.selectMethods(targetClass,(MethodIntrospector.MetadataLookup<Set<KafkaListener>>) method -> {Set<KafkaListener> listenerMethods = findListenerAnnotations(method);return (!listenerMethods.isEmpty() ? listenerMethods : null);});if (hasClassLevelListeners) {Set<Method> methodsWithHandler = MethodIntrospector.selectMethods(targetClass,(ReflectionUtils.MethodFilter) method ->AnnotationUtils.findAnnotation(method, KafkaHandler.class) != null);multiMethods.addAll(methodsWithHandler);}if (annotatedMethods.isEmpty()) {this.nonAnnotatedClasses.add(bean.getClass());if (this.logger.isTraceEnabled()) {this.logger.trace("No @KafkaListener annotations found on bean type: " + bean.getClass());}}else {// Non-empty set of methodsfor (Map.Entry<Method, Set<KafkaListener>> entry : annotatedMethods.entrySet()) {Method method = entry.getKey();for (KafkaListener listener : entry.getValue()) {processKafkaListener(listener, method, bean, beanName);}}if (this.logger.isDebugEnabled()) {this.logger.debug(annotatedMethods.size() + " @KafkaListener methods processed on bean '"+ beanName + "': " + annotatedMethods);}}if (hasClassLevelListeners) {processMultiMethodListeners(classLevelListeners, multiMethods, bean, beanName);}}return bean;}
}
KafkaListenerAnnotationBeanPostProcessor实现了BeanPostProcessor, Ordered, BeanFactoryAware, SmartInitializingSingleton接口,其getOrder返回LOWEST_PRECEDENCE
其afterSingletonsInstantiated方法(SmartInitializingSingleton接口
)首先获取KafkaListenerConfigurer,然后设置configureKafkaListeners为registrar,最后是执行registrar.afterPropertiesSet()
其postProcessAfterInitialization方法(BeanPostProcessor接口
)会收集标注KafkaListener的bean的方法,然后针对每个方法执行processKafkaListener
processKafkaListener
protected void processKafkaListener(KafkaListener kafkaListener, Method method, Object bean, String beanName) {Method methodToUse = checkProxy(method, bean);MethodKafkaListenerEndpoint<K, V> endpoint = new MethodKafkaListenerEndpoint<>();endpoint.setMethod(methodToUse);processListener(endpoint, kafkaListener, bean, methodToUse, beanName);}protected void processListener(MethodKafkaListenerEndpoint<?, ?> endpoint, KafkaListener kafkaListener,Object bean, Object adminTarget, String beanName) {String beanRef = kafkaListener.beanRef();if (StringUtils.hasText(beanRef)) {this.listenerScope.addListener(beanRef, bean);}endpoint.setBean(bean);endpoint.setMessageHandlerMethodFactory(this.messageHandlerMethodFactory);endpoint.setId(getEndpointId(kafkaListener));endpoint.setGroupId(getEndpointGroupId(kafkaListener, endpoint.getId()));endpoint.setTopicPartitions(resolveTopicPartitions(kafkaListener));endpoint.setTopics(resolveTopics(kafkaListener));endpoint.setTopicPattern(resolvePattern(kafkaListener));endpoint.setClientIdPrefix(resolveExpressionAsString(kafkaListener.clientIdPrefix(), "clientIdPrefix"));String group = kafkaListener.containerGroup();if (StringUtils.hasText(group)) {Object resolvedGroup = resolveExpression(group);if (resolvedGroup instanceof String) {endpoint.setGroup((String) resolvedGroup);}}String concurrency = kafkaListener.concurrency();if (StringUtils.hasText(concurrency)) {endpoint.setConcurrency(resolveExpressionAsInteger(concurrency, "concurrency"));}String autoStartup = kafkaListener.autoStartup();if (StringUtils.hasText(autoStartup)) {endpoint.setAutoStartup(resolveExpressionAsBoolean(autoStartup, "autoStartup"));}resolveKafkaProperties(endpoint, kafkaListener.properties());KafkaListenerContainerFactory<?> factory = null;String containerFactoryBeanName = resolve(kafkaListener.containerFactory());if (StringUtils.hasText(containerFactoryBeanName)) {Assert.state(this.beanFactory != null, "BeanFactory must be set to obtain container factory by bean name");try {factory = this.beanFactory.getBean(containerFactoryBeanName, KafkaListenerContainerFactory.class);}catch (NoSuchBeanDefinitionException ex) {throw new BeanInitializationException("Could not register Kafka listener endpoint on [" + adminTarget+ "] for bean " + beanName + ", no " + KafkaListenerContainerFactory.class.getSimpleName()+ " with id '" + containerFactoryBeanName + "' was found in the application context", ex);}}endpoint.setBeanFactory(this.beanFactory);String errorHandlerBeanName = resolveExpressionAsString(kafkaListener.errorHandler(), "errorHandler");if (StringUtils.hasText(errorHandlerBeanName)) {endpoint.setErrorHandler(this.beanFactory.getBean(errorHandlerBeanName, KafkaListenerErrorHandler.class));}this.registrar.registerEndpoint(endpoint, factory);if (StringUtils.hasText(beanRef)) {this.listenerScope.removeListener(beanRef);}}
processKafkaListener方法将method转换为MethodKafkaListenerEndpoint,然后执行processListener方法,它主要是将KafkaListener注解的信息填充到MethodKafkaListenerEndpoint上,确定KafkaListenerContainerFactory,最后执行registrar.registerEndpoint(endpoint, factory)
registrar.registerEndpoint
org/springframework/kafka/config/KafkaListenerEndpointRegistrar.java
/*** Register a new {@link KafkaListenerEndpoint} alongside the* {@link KafkaListenerContainerFactory} to use to create the underlying container.* <p>The {@code factory} may be {@code null} if the default factory has to be* used for that endpoint.* @param endpoint the {@link KafkaListenerEndpoint} instance to register.* @param factory the {@link KafkaListenerContainerFactory} to use.*/public void registerEndpoint(KafkaListenerEndpoint endpoint, KafkaListenerContainerFactory<?> factory) {Assert.notNull(endpoint, "Endpoint must be set");Assert.hasText(endpoint.getId(), "Endpoint id must be set");// Factory may be null, we defer the resolution right before actually creating the containerKafkaListenerEndpointDescriptor descriptor = new KafkaListenerEndpointDescriptor(endpoint, factory);synchronized (this.endpointDescriptors) {if (this.startImmediately) { // Register and start immediatelythis.endpointRegistry.registerListenerContainer(descriptor.endpoint,resolveContainerFactory(descriptor), true);}else {this.endpointDescriptors.add(descriptor);}}}
KafkaListenerEndpointRegistrar的registerEndpoint会创建KafkaListenerEndpointDescriptor,然后执行endpointRegistry.registerListenerContainer
endpointRegistry.registerListenerContainer
org/springframework/kafka/config/KafkaListenerEndpointRegistry.java
public void registerListenerContainer(KafkaListenerEndpoint endpoint, KafkaListenerContainerFactory<?> factory,boolean startImmediately) {Assert.notNull(endpoint, "Endpoint must not be null");Assert.notNull(factory, "Factory must not be null");String id = endpoint.getId();Assert.hasText(id, "Endpoint id must not be empty");synchronized (this.listenerContainers) {Assert.state(!this.listenerContainers.containsKey(id),"Another endpoint is already registered with id '" + id + "'");MessageListenerContainer container = createListenerContainer(endpoint, factory);this.listenerContainers.put(id, container);if (StringUtils.hasText(endpoint.getGroup()) && this.applicationContext != null) {List<MessageListenerContainer> containerGroup;if (this.applicationContext.containsBean(endpoint.getGroup())) {containerGroup = this.applicationContext.getBean(endpoint.getGroup(), List.class);}else {containerGroup = new ArrayList<MessageListenerContainer>();this.applicationContext.getBeanFactory().registerSingleton(endpoint.getGroup(), containerGroup);}containerGroup.add(container);}if (startImmediately) {startIfNecessary(container);}}}/*** Start the specified {@link MessageListenerContainer} if it should be started* on startup.* @param listenerContainer the listener container to start.* @see MessageListenerContainer#isAutoStartup()*/private void startIfNecessary(MessageListenerContainer listenerContainer) {if (this.contextRefreshed || listenerContainer.isAutoStartup()) {listenerContainer.start();}}
KafkaListenerEndpointRegistry的registerListenerContainer方法会根据endpoint和factory来创建MessageListenerContainer,然后放入到listenerContainers中,对于startImmediately的会执行startIfNecessary,它主要是执行listenerContainer.start()
MessageListenerContainer
org/springframework/kafka/listener/MessageListenerContainer.java
public interface MessageListenerContainer extends SmartLifecycle {void setupMessageListener(Object messageListener);Map<String, Map<MetricName, ? extends Metric>> metrics();default ContainerProperties getContainerProperties() {throw new UnsupportedOperationException("This container doesn't support retrieving its properties");}default Collection<TopicPartition> getAssignedPartitions() {throw new UnsupportedOperationException("This container doesn't support retrieving its assigned partitions");}default void pause() {throw new UnsupportedOperationException("This container doesn't support pause");}default void resume() {throw new UnsupportedOperationException("This container doesn't support resume");}default boolean isPauseRequested() {throw new UnsupportedOperationException("This container doesn't support pause/resume");}default boolean isContainerPaused() {throw new UnsupportedOperationException("This container doesn't support pause/resume");}default void setAutoStartup(boolean autoStartup) {// empty}default String getGroupId() {throw new UnsupportedOperationException("This container does not support retrieving the group id");}@Nullabledefault String getListenerId() {throw new UnsupportedOperationException("This container does not support retrieving the listener id");}
}
MessageListenerContainer继承了SmartLifecycle接口,它有一个泛型接口为GenericMessageListenerContainer,后者有一个抽象类为AbstractMessageListenerContainer,然后它有两个子类,分别是KafkaMessageListenerContainer与ConcurrentMessageListenerContainer
AbstractMessageListenerContainer
public abstract class AbstractMessageListenerContainer<K, V>implements GenericMessageListenerContainer<K, V>, BeanNameAware, ApplicationEventPublisherAware {@Overridepublic final void start() {checkGroupId();synchronized (this.lifecycleMonitor) {if (!isRunning()) {Assert.isTrue(this.containerProperties.getMessageListener() instanceof GenericMessageListener,() -> "A " + GenericMessageListener.class.getName() + " implementation must be provided");doStart();}}}@Overridepublic final void stop() {synchronized (this.lifecycleMonitor) {if (isRunning()) {final CountDownLatch latch = new CountDownLatch(1);doStop(latch::countDown);try {latch.await(this.containerProperties.getShutdownTimeout(), TimeUnit.MILLISECONDS); // NOSONARpublishContainerStoppedEvent();}catch (InterruptedException e) {Thread.currentThread().interrupt();}}}}//......
}
AbstractMessageListenerContainer的start方法会回调子类的doStart方法,其stop方法会回调子类的doStop方法
KafkaMessageListenerContainer
org/springframework/kafka/listener/KafkaMessageListenerContainer.java
public class KafkaMessageListenerContainer<K, V> // NOSONAR comment densityextends AbstractMessageListenerContainer<K, V> {@Overrideprotected void doStart() {if (isRunning()) {return;}if (this.clientIdSuffix == null) { // stand-alone containercheckTopics();}ContainerProperties containerProperties = getContainerProperties();checkAckMode(containerProperties);Object messageListener = containerProperties.getMessageListener();Assert.state(messageListener != null, "A MessageListener is required");if (containerProperties.getConsumerTaskExecutor() == null) {SimpleAsyncTaskExecutor consumerExecutor = new SimpleAsyncTaskExecutor((getBeanName() == null ? "" : getBeanName()) + "-C-");containerProperties.setConsumerTaskExecutor(consumerExecutor);}Assert.state(messageListener instanceof GenericMessageListener, "Listener must be a GenericListener");GenericMessageListener<?> listener = (GenericMessageListener<?>) messageListener;ListenerType listenerType = deteremineListenerType(listener);this.listenerConsumer = new ListenerConsumer(listener, listenerType);setRunning(true);this.listenerConsumerFuture = containerProperties.getConsumerTaskExecutor().submitListenable(this.listenerConsumer);}//......
}
KafkaMessageListenerContainer的doStart方法会获取到messageListener,然后创建ListenerConsumer,最后提交到线程池中执行
ConcurrentMessageListenerContainer
org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java
public class ConcurrentMessageListenerContainer<K, V> extends AbstractMessageListenerContainer<K, V> {@Overrideprotected void doStart() {if (!isRunning()) {checkTopics();ContainerProperties containerProperties = getContainerProperties();TopicPartitionInitialOffset[] topicPartitions = containerProperties.getTopicPartitions();if (topicPartitions != null && this.concurrency > topicPartitions.length) {this.logger.warn("When specific partitions are provided, the concurrency must be less than or "+ "equal to the number of partitions; reduced from " + this.concurrency + " to "+ topicPartitions.length);this.concurrency = topicPartitions.length;}setRunning(true);for (int i = 0; i < this.concurrency; i++) {KafkaMessageListenerContainer<K, V> container;if (topicPartitions == null) {container = new KafkaMessageListenerContainer<>(this, this.consumerFactory, containerProperties);}else {container = new KafkaMessageListenerContainer<>(this, this.consumerFactory,containerProperties, partitionSubset(containerProperties, i));}String beanName = getBeanName();container.setBeanName((beanName != null ? beanName : "consumer") + "-" + i);if (getApplicationEventPublisher() != null) {container.setApplicationEventPublisher(getApplicationEventPublisher());}container.setClientIdSuffix("-" + i);container.setGenericErrorHandler(getGenericErrorHandler());container.setAfterRollbackProcessor(getAfterRollbackProcessor());container.setRecordInterceptor(getRecordInterceptor());container.setEmergencyStop(() -> {stop(() -> {// NOSONAR});publishContainerStoppedEvent();});if (isPaused()) {container.pause();}container.start();this.containers.add(container);}}}//......
}
ConcurrentMessageListenerContainer的doStart会根据concurrency值来创建对应的KafkaMessageListenerContainer,然后执行其start方法
ListenerConsumer
org/springframework/kafka/listener/KafkaMessageListenerContainer.java
private final class ListenerConsumer implements SchedulingAwareRunnable, ConsumerSeekCallback {@Overridepublic void run() {this.consumerThread = Thread.currentThread();if (this.genericListener instanceof ConsumerSeekAware) {((ConsumerSeekAware) this.genericListener).registerSeekCallback(this);}if (this.transactionManager != null) {ProducerFactoryUtils.setConsumerGroupId(this.consumerGroupId);}this.count = 0;this.last = System.currentTimeMillis();initAsignedPartitions();while (isRunning()) {try {pollAndInvoke();}catch (@SuppressWarnings(UNUSED) WakeupException e) {// Ignore, we're stopping or applying immediate foreign acks}catch (NoOffsetForPartitionException nofpe) {this.fatalError = true;ListenerConsumer.this.logger.error("No offset and no reset policy", nofpe);break;}catch (Exception e) {handleConsumerException(e);}catch (Error e) { // NOSONAR - rethrownRunnable runnable = KafkaMessageListenerContainer.this.emergencyStop;if (runnable != null) {runnable.run();}this.logger.error("Stopping container due to an Error", e);wrapUp();throw e;}}wrapUp();}protected void pollAndInvoke() {if (!this.autoCommit && !this.isRecordAck) {processCommits();}processSeeks();checkPaused();ConsumerRecords<K, V> records = this.consumer.poll(this.pollTimeout);this.lastPoll = System.currentTimeMillis();checkResumed();debugRecords(records);if (records != null && records.count() > 0) {if (this.containerProperties.getIdleEventInterval() != null) {this.lastReceive = System.currentTimeMillis();}invokeListener(records);}else {checkIdle();}}private void invokeListener(final ConsumerRecords<K, V> records) {if (this.isBatchListener) {invokeBatchListener(records);}else {invokeRecordListener(records);}}private void doInvokeBatchOnMessage(final ConsumerRecords<K, V> records,List<ConsumerRecord<K, V>> recordList) {switch (this.listenerType) {case ACKNOWLEDGING_CONSUMER_AWARE:this.batchListener.onMessage(recordList,this.isAnyManualAck? new ConsumerBatchAcknowledgment(records): null, this.consumer);break;case ACKNOWLEDGING:this.batchListener.onMessage(recordList,this.isAnyManualAck? new ConsumerBatchAcknowledgment(records): null);break;case CONSUMER_AWARE:this.batchListener.onMessage(recordList, this.consumer);break;case SIMPLE:this.batchListener.onMessage(recordList);break;}}private void doInvokeOnMessage(final ConsumerRecord<K, V> recordArg) {ConsumerRecord<K, V> record = recordArg;if (this.recordInterceptor != null) {record = this.recordInterceptor.intercept(record);}if (record == null) {if (this.logger.isDebugEnabled()) {this.logger.debug("RecordInterceptor returned null, skipping: " + recordArg);}}else {switch (this.listenerType) {case ACKNOWLEDGING_CONSUMER_AWARE:this.listener.onMessage(record,this.isAnyManualAck? new ConsumerAcknowledgment(record): null, this.consumer);break;case CONSUMER_AWARE:this.listener.onMessage(record, this.consumer);break;case ACKNOWLEDGING:this.listener.onMessage(record,this.isAnyManualAck? new ConsumerAcknowledgment(record): null);break;case SIMPLE:this.listener.onMessage(record);break;}}} //......
}
ListenerConsumer实现了org.springframework.scheduling.SchedulingAwareRunnable接口(
它继承了Runnable接口
)以及org.springframework.kafka.listener.ConsumerSeekCallback接口
其run方法主要是执行initAsignedPartitions,然后循环执行pollAndInvoke,对于NoOffsetForPartitionException则跳出异常,对于其他Exception则执行handleConsumerException,对于Error执行emergencyStop与wrapUp方法
pollAndInvoke方法主要是执行consumer.poll(),然后通过invokeListener(records)回调,最后是通过doInvokeBatchOnMessage、doInvokeOnMessage去回调listener.onMessage方法
小结
KafkaListenerAnnotationBeanPostProcessor主要是收集标注KafkaListener的bean的方法,然后针对每个方法执行processKafkaListener,processKafkaListener方法将method转换为MethodKafkaListenerEndpoint,执行registrar.registerEndpoint(endpoint, factory)
KafkaListenerEndpointRegistry的registerListenerContainer方法会根据endpoint和factory来创建MessageListenerContainer,然后放入到listenerContainers中,对于startImmediately的会执行startIfNecessary,它主要是执行listenerContainer.start()
MessageListenerContainer有两个主要的实现类分别是KafkaMessageListenerContainer与ConcurrentMessageListenerContainer,后者的start方法主要是根据concurrency创建对应数量的KafkaMessageListenerContainer,最后都是执行KafkaMessageListenerContainer的start方法,它会创建ListenerConsumer,最后提交到线程池中执行;ListenerConsumer主要是执行pollAndInvoke,拉取消息,然后回到listener的onMessage方法
整体的链路就是KafkaListenerAnnotationBeanPostProcessor --> KafkaListenerEndpointRegistry --> MessageListenerContainer --> GenericMessageListener.onMessage
相关文章:
聊聊KafkaListener的实现机制
序 本文只要研究一下KafkaListener的实现机制 KafkaListener org/springframework/kafka/annotation/KafkaListener.java Target({ ElementType.TYPE, ElementType.METHOD, ElementType.ANNOTATION_TYPE }) Retention(RetentionPolicy.RUNTIME) MessageMapping Documented …...
Golang洗牌算法(Golang乱序算法)
Golang 洗牌算法(乱序算法);需求背景:从一个文件下下读取所有文件,获取他们的名字, 将名字乱序排序,按着乱序后的序列,通过名字去找到文件,再上传,以达到上传…...

SpringBoot 源码分析(三) 监听器分析以及属性文件加载分析
前言 在创建SpringBoot项目的时候会在对应的application.properties或者application.yml文件中添加对应的属性信息,这些属性文件是什么时候被加载的?如果要实现自定义的属性文件怎么来实现?在讲属性加载之前先讲下监听器分析。 一、监听器分…...
记录nvm use node.js版本失败,出现报错: exit status 1: ��û���㹻��Ȩ��ִ�д˲�����
使用管理员权限运行cmd,再使用nvm use node.js版本号 参考: nvm use (node版本号)时报错: exit status 1: �����㹻��Ȩ��ִ…...

【蓝牙协议】简介:蓝牙芯片、蓝牙协议架构
文章目录 蓝牙芯片架构另一个视角由下到上看:Controller-->Host由上到下看:Host-->Controller 蓝牙协议架构视角HW层——蓝牙芯片层Transport——数据传输层HOST——协议层 总结 参考:https://zhuanlan.zhihu.com/p/585248998 参考&…...
【深度学习】
什么是深度学习? 感知器 为了实现模拟人类的学习,科学家们首先设计了构成神经网络的基本结构神经元(感知器模型),然后再由大量的神经元构成复杂的,能够实现各种功能的神经网络。 这种模式和超能陆战队中的…...

centos启动tomcat 并指定jdk 版本
在tomcat的catalina.sh文件手动设置JAVA_HOME变量即可 例如: 前提是文件存在 保存配置重新启动tomcat...
day37(事件轮询机制 ajaxGet执行步骤与案例(五个步骤) ajax属性 PHP返回JSON对象(两种))
一.事件轮询机制 1. 无论同步还是异步代码都要经过主线程编译,同步代码开始排在执行栈(主线程)上,异步代码开 始存放在任务队列中 2. 主线程优先执行同步代码,同步代码必须前一行执行完,后一行才能执行;当异步代码…...

Flume基本使用--mysql数据输出
MySQL数据输出 在MySQL中建立数据库school,在数据库中建立表student。SQL语句如下: create database school; use school; create table student(id int not null,name varchar(40),age int,grade int,primary key(id) ); 请使用Flume实时捕…...

MySQL——EXPLAIN用法详解
EXPLAIN是MySQL官方提供的sql分析的工具之一,可以用于模拟优化器执行sql查询语句,从而知道MySQL是如何处理sql语句。EXPLAIN主要用于分析查询语句或表结构的性能瓶颈。 以下是基于MySQL5.7.19版本进行分析的,不同版本之间略有差异。 1、EXP…...

69 划分字母区间
划分字母区间 题解1 贪心1(方法略笨,性能很差)题解2 贪心2(参考标答) 给你一个字符串 s 。我们要把这个字符串划分为尽可能多的片段,同一字母最多出现在一个片段中。 注意,划分结果需要满足&am…...
文件上传漏洞(1), 文件上传绕过原理
文件上传漏洞 一, 前端校验上传文件 添加 Javascript 代码,然后在 form 表单中 添加 onsubmit"returb checkFile()" <script>function checkFile() {// var file document.getElementsByName(photo)[0].value;var file document.getElementByI…...
【ARM 嵌入式 C 入门及渐进 10 -- 冒泡排序 选择排序 插入排序 快速排序 归并排序 堆排序 比较介绍】
文章目录 排序算法小结排序算法C实现 排序算法小结 C语言中常用的排序算法包括冒泡排序、选择排序、插入排序、快速排序、归并排序、堆排序。下面我们来一一介绍: 冒泡排序(Bubble Sort):冒泡排序是通过比较相邻元素的大小进行排…...

虹科 | 解决方案 | 汽车示波器 学校教学方案
虹科Pico汽车示波器是基于PC的设备,特别适用于大课堂的教学、备课以及与师生的互动交流。老师展现讲解波形数据,让学生直观形象地理解汽车的工作原理 高效备课 课前实测,采集波形数据,轻松截图与标注,制作优美的课件&…...
广播和组播(多播)
广播 概述 广播(broadcast)是指封包在计算机网络中传输时,目的地址为网络中所有设备的一种传输方式。实际上,这里所说的“所有设备”也是限定在一个范围之中,称为“广播域”。并非所有的计算机网络都支持广播…...

【Linux】gdb调试
目录 进入调试查看代码运行代码断点打断点查断点删断点从一个断点转跳至下一个断点保留断点但不会运行该断点 退出调试逐过程逐语句监视跳转至指定行运行结束当前函数 进入调试 指令:gdb 【可执行文件】: 查看代码 :l 【第几行】如果输入指…...
MySQL创建函数及其使用
MySQL创建函数及其使用 一、MySQL 创建函数二、示例 一、MySQL 创建函数 MySQL 函数是一种可重用的代码块,可以接受输入参数并返回值。你可以在 MySQL 中创建各种类型的函数,包括系统函数、用户定义函数和存储过程。在此处,我们将重点关注用…...

大数据-Storm流式框架(四)---storm容错机制
1、集群节点宕机 Nimbus服务器 硬件 单点故障?可以搭建HA jStorm搭建 nimbus的HA nimbus的信息存储到zookeeper中,只要下游没问题(进程退出)nimbus退出就不会有问题, 如果在nimbus宕机,也不能提交…...

SpringBoot项目把Mysql从5.7升级到8.0
首先你需要把之前的库导入到mysql库导入到8.0的新库中。(导入的时候会报错我是通过navcat备份恢复的) 1、项目中需要修改pom文件的依赖 mysql 和 jdbc <dependency><groupId>mysql</groupId><artifactId>mysql-connector-java&…...

RK3568-适配at24c04模块
将at24c04模块连接到开发板i2c2总线上 i2ctool查看i2c2总线上都有哪些设备 UU表示设备地址的从设备被驱动占用,卸载对应的驱动后,UU就会变成从设备地址。at24c04模块设备地址 0x50和0x51是at24c04模块i2c芯片的设备地址。这个从芯片手册上也可以得知。A0 A1 A2表示的是模块对…...

.Net框架,除了EF还有很多很多......
文章目录 1. 引言2. Dapper2.1 概述与设计原理2.2 核心功能与代码示例基本查询多映射查询存储过程调用 2.3 性能优化原理2.4 适用场景 3. NHibernate3.1 概述与架构设计3.2 映射配置示例Fluent映射XML映射 3.3 查询示例HQL查询Criteria APILINQ提供程序 3.4 高级特性3.5 适用场…...

SCAU期末笔记 - 数据分析与数据挖掘题库解析
这门怎么题库答案不全啊日 来简单学一下子来 一、选择题(可多选) 将原始数据进行集成、变换、维度规约、数值规约是在以下哪个步骤的任务?(C) A. 频繁模式挖掘 B.分类和预测 C.数据预处理 D.数据流挖掘 A. 频繁模式挖掘:专注于发现数据中…...

IoT/HCIP实验-3/LiteOS操作系统内核实验(任务、内存、信号量、CMSIS..)
文章目录 概述HelloWorld 工程C/C配置编译器主配置Makefile脚本烧录器主配置运行结果程序调用栈 任务管理实验实验结果osal 系统适配层osal_task_create 其他实验实验源码内存管理实验互斥锁实验信号量实验 CMISIS接口实验还是得JlINKCMSIS 简介LiteOS->CMSIS任务间消息交互…...

HashMap中的put方法执行流程(流程图)
1 put操作整体流程 HashMap 的 put 操作是其最核心的功能之一。在 JDK 1.8 及以后版本中,其主要逻辑封装在 putVal 这个内部方法中。整个过程大致如下: 初始判断与哈希计算: 首先,putVal 方法会检查当前的 table(也就…...
LeetCode - 199. 二叉树的右视图
题目 199. 二叉树的右视图 - 力扣(LeetCode) 思路 右视图是指从树的右侧看,对于每一层,只能看到该层最右边的节点。实现思路是: 使用深度优先搜索(DFS)按照"根-右-左"的顺序遍历树记录每个节点的深度对于…...

HDFS分布式存储 zookeeper
hadoop介绍 狭义上hadoop是指apache的一款开源软件 用java语言实现开源框架,允许使用简单的变成模型跨计算机对大型集群进行分布式处理(1.海量的数据存储 2.海量数据的计算)Hadoop核心组件 hdfs(分布式文件存储系统)&a…...

PHP 8.5 即将发布:管道操作符、强力调试
前不久,PHP宣布了即将在 2025 年 11 月 20 日 正式发布的 PHP 8.5!作为 PHP 语言的又一次重要迭代,PHP 8.5 承诺带来一系列旨在提升代码可读性、健壮性以及开发者效率的改进。而更令人兴奋的是,借助强大的本地开发环境 ServBay&am…...
uniapp 实现腾讯云IM群文件上传下载功能
UniApp 集成腾讯云IM实现群文件上传下载功能全攻略 一、功能背景与技术选型 在团队协作场景中,群文件共享是核心需求之一。本文将介绍如何基于腾讯云IMCOS,在uniapp中实现: 群内文件上传/下载文件元数据管理下载进度追踪跨平台文件预览 二…...
鸿蒙(HarmonyOS5)实现跳一跳小游戏
下面我将介绍如何使用鸿蒙的ArkUI框架,实现一个简单的跳一跳小游戏。 1. 项目结构 src/main/ets/ ├── MainAbility │ ├── pages │ │ ├── Index.ets // 主页面 │ │ └── GamePage.ets // 游戏页面 │ └── model │ …...

Qt的学习(一)
1.什么是Qt Qt特指用来进行桌面应用开发(电脑上写的程序)涉及到的一套技术Qt无法开发网页前端,也不能开发移动应用。 客户端开发的重要任务:编写和用户交互的界面。一般来说和用户交互的界面,有两种典型风格&…...