外贸做企业什么网站建设申请付费网站

外贸做企业什么网站建设,申请付费网站,小程序推广员好做吗,网站首页收录没有了动态消费的使用场景首先#xff0c;什么是动态消费#xff1f; 简单来讲#xff0c;就是Spring Kafka提供了安全地在运行时调整消费状态的实现#xff0c;可以随时调整消费者的消费状态。 比如暂时停止正在消费消息的消费者的消费#xff0c;等到合适的时机再重新从中断的…动态消费的使用场景首先什么是动态消费 简单来讲就是Spring Kafka提供了安全地在运行时调整消费状态的实现可以随时调整消费者的消费状态。 比如暂时停止正在消费消息的消费者的消费等到合适的时机再重新从中断的地方开始消费。 再比如关闭消费者线程不再消费在执行某些操作如修改消费者参数之后再重新创建消费者并开始消费。下面详细说说。开关降级当一个新功能发布之后还没有正式使用消费线程就没有创建的必要此时KafkaListener注解参数autoStartup false就可以实现。 如果某个功能需要降级也可以销毁消费线程不再注册这个消费者。延迟消费开始消费前需要进行自检、数据准备等操作。 有时候消费消息需要提前检查某些配置或者数据是否已经存在或者当服务启动之后需要异步完成一些资源的初始化才能正常消费。此时可以等待完成准备之后再执行start()方法创建消费线程。限流背压在高峰期由消费者端控制防止系统崩溃使用pause() 暂停拉取消息、resume()恢复拉取消息。参数调整不停机修改消费配置如 groupId、topic使用stop() / start() 重新创建消费者会触发消费者重平衡会重新建立与 Kafka 的连接。SpringKafka消费的实现机制在实现Spring Kafka动态调整消费之前需要先了解Spring Kafka的消费者是如何创建的KafkaListener 用于声明一个方法为Kafka消息监听器。 Spring Kafka 会在应用启动时自动完成以下工作扫描并解析 KafkaListener 注解创建对应的监听端点KafkaListenerEndpoint将端点注册到 KafkaListenerEndpointRegistry使用 KafkaListenerContainerFactory 创建监听容器启动监听容器开始消费消息。详细描述如下Java KafkaListener ↓ 【后处理器PostProcessor】 Spring在启动时会加载一个后处理器[KafkaListenerAnnotationBeanPostProcessor] 负责扫描所有Spring Bean查找带有 KafkaListener 的方法。 ↓ 【监听端点ListenerEndpoint】 创建监听端点[MethodKafkaListenerEndpoint] 负责定义要监听的主题消费组监听方法消息反序列化类型容器工厂名称等。 ↓ 【全局注册表EndpointRegistry】 Processor将Endpoint注册到全局注册表[KafkaListenerEndpointRegistry] 保存所有监听端点管理监听容器的生命周期启动、停止、暂停、恢复提供运行时访问接口。 ↓ 【容器工厂ContainerFactory】 创建监听容器工厂[KafkaListenerContainerFactory] 注册表使用工厂创建实际的监听容器 ↓ 【监听容器ListenerContainer】 拉取消息并调用监听方法[ConcurrentMessageListenerContainer] 根据配置autoStartuptrue自动启动所有监听容器 ↓ KafkaConsumer通过上面的描述我们知道KafkaListenerEndpointRegistry注册表管理者所有容器MessageListenerContainer 接口的实现类保存着所有消费者容器我们可以从消费者容器中获取到我们的消费者。那么MessageListenerContainer提供了哪些方法呢 主要是下面4个pause()暂停拉取消息不释放资源resume()恢复拉取消息不释放资源stop()停止容器可重新启动不释放资源start()启动容器不释放资源这4个方法可以分为两组分别是pause() / resume()stop() / start()如何进行选择下面详细讲解一下它们的差异。pause() / resume()这两个方法是对消费者的轻量级控制暂停或恢复消息拉取不影响容器生命周期。适用场景是临时暂停消费例如下游系统压力过大或实现“背压”机制或动态限流或批量处理控制。pause()暂停拉取消息但保持与 Kafka Broker 的连接不会提交偏移量也不会关闭线程resume()恢复拉取消息继续消费。pause()之后容器仍在运行状态isRunning() true但isContainerPaused() true消费线程仍存在不会触发消费者重平衡恢复后从上次暂停的偏移继续消费。stop() / start()重级控制停止或启动整个监听容器会释放或重建资源。适合场景应用启动时延迟启动消费动态启停某个监听器部署或维护期间临时关闭消费切换消费配置如 groupId、topic。stop()停止容器运行关闭所有 KafkaConsumer释放线程池和资源容器状态变为 isRunning() false。start()重新创建 KafkaConsumer重新订阅主题启动消费线程容器状态变为 isRunning() true。Consumer动态消费实现案例消费者Consumer注解:iniKafkaListener( topics ${spring.kafka.multiple.consumer.mybiz.topics}, concurrency ${spring.kafka.multiple.consumer.mybiz.concurrency}, containerFactory mybizConsumer, id mybizListenerId, // 容器id autoStartup false // 是否自动注册,是true,否false,默认是 )服务启动之后由于 autoStartup false 不会注册消费者。可以通过下面的管理器调整消费状态typescript Service public class KafkaListenerAdjuster { Autowired private KafkaListenerEndpointRegistry registry; // 暂停消费 public void pause(String listenerId) { MessageListenerContainer container registry.getListenerContainer(listenerId); if (container ! null container.isRunning()) { container.pause(); } } // 恢复消费 public void resume(String listenerId) { MessageListenerContainer container registry.getListenerContainer(listenerId); if (container ! null) { container.resume(); } } // 停止消费 public void stop(String listenerId) { MessageListenerContainer container registry.getListenerContainer(listenerId); if (container ! null) { container.stop(); } } // 启动消费 public void start(String listenerId) { MessageListenerContainer container registry.getListenerContainer(listenerId); if (container ! null !container.isRunning()) { container.start(); } } // 动态调整并发数 public void changeConcurrency(String listenerId, int concurrency) { ConcurrentMessageListenerContainer?, ? container (ConcurrentMessageListenerContainer?, ?) registry.getListenerContainer(listenerId); if (container ! null) { container.setConcurrency(concurrency); } } }总结以上的实现在实际项目开发和生产环境可以和分布式配置管理框架或注册中心Spring Cloud Config、Nacos、ZooKeeper、Apollo、Consul等结合通过事件如SpringEvent等进行动态调整。在分布式系统或容器中需要确保每台机器的消费状态都得到了调整。除了动态调整消费状态之外还可以 Spring Kafka 源码的基础之上实现其他能力的动态调整如增加自定义过滤器可以实现消息的过滤消费或者在消息中增加灰度标识等。