服务老重启线程数暴增

背景

  生产服务部署在k8s 集群上，每个pod分配最大4G内存，运行一段时间后，pod会自动重启。猜测可能原因1. 节点重启了或者集群资源不足，导致pod集群内迁移2. 内存溢出，OOM问题，被杀掉3. 假死，被杀掉，导致重启

集群情况

查看集群资源，总体内存没有达到集群80%限制，节点未重启过
查看集群节点资源，节点资源仍有余量
结论，不太可能是“集群资源不足，导致pod集群内迁移”，先排除。

pod服务情况

app.jar，依赖lib文件夹，及日志，
查看日志文件总体大小，发现日志很大，手动删除一些日志，发现该pod占用集群的内存会相应减少。
结论，日志文件没有落盘占了集群内存资源，导致重启或者群集内漂移
优化调整方案：
1日志落盘，将日志文件落盘到节点目录上，但随之而来的是pod重建销毁对应日志的管理。
2减少日志输出，指输出必要日志，及调高特定类日志级别，或干脆关闭特定类的日志输出
3不输出日志文件，采集集群日志
这里先采取第二种方案减少日志输出。

额外问题

1.服务内存与预期不对
集群设置客户服务环境变量
- name: JAVA_TOOL_OPTIONS
value: “-Xmx3g -Xms2g”
使用arthus，jvm 命令查看，实际内存初始化为1g，最大2g
查看dockerfile发现问题所在，将内存限制删掉，使用集群环境变量JAVA_TOOL_OPTIONS
CMD [“java”,“-Duser.timezone=Asia/Shanghai”,“-Djava.security.egd=file:/dev/./urandom”,“-jar”,“-Xms1024m”,“-Xmx2044m”,“.app.jar”]

2.线程数有暴增变化
在问题监控中，发现线程数有暴增变化

• 业务场景中存在聚合逻辑
  搜索客户信息，查到客户信息的各种id，标签、客户群、好友、基础信息、扩展信息等
  按id聚合客户信息，forkJoinPool线程池join 导致，可参考文档
  https://juejin.cn/post/6844904196404150286

这里暂时不动，可以按需求拆分进行只查询客户基本数据。

• mq消费者设置数量范围过大
  代码中mq并发消费者初始值5，最大20

    SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();factory.setConnectionFactory(connectionFactory);factory.setPrefetchCount(5); // 每个消费者每次监听可拉取消息的数量factory.setMaxConcurrentConsumers(20); // 并发消费者最大值factory.setConcurrentConsumers(5); // 并发消费者初始值factory.setDefaultRequeueRejected(false); // 消息处理异常时是否丢弃消息，设置true有死循环风险factory.setAcknowledgeMode(AcknowledgeMode.AUTO); // 自动应答模式，消息接收处理异常情况下丢失消息return factory;

消息消费使用处理逻辑代码

@RabbitListener(bindings =@QueueBinding(value =@Queue(value = QUEUE,ignoreDeclarationExceptions = "true"),exchange =@Exchange(value = EVENT_EXCHANGE,ignoreDeclarationExceptions = "true",type = ExchangeTypes.TOPIC),key = ROUTING_KEY))@Async("mqExecutor")public void receiveMessage(MqModel<Object> model) {}@Beanpublic Executor mqExecutor() {ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();executor.setCorePoolSize(4);executor.setMaxPoolSize(20);executor.setQueueCapacity(100);executor.setThreadNamePrefix("mqExecutor-");// rejection-policy：当pool已经达到max size的时候，如何处理新任务// CALLER_RUNS：不在新线程中执行任务，而是有调用者所在的线程来执行executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());executor.initialize();return executor;}

代码中消息消费，是用线程池异步处理，其目的是限制线程数及消息处理时间过长

队列消息增多，并发消费者数量可以从5个增加到20个，而真正消费逻辑使用线程池处理逻辑，看似限制了线程数量，其实不然。mqExecutor线程池的拒绝执行策略为CALLER_RUNS，即调用者所在的线程来执行。也就说mqExecutor线程池并没有限制住线程数量，而是由队列并发消费者线程，即调用处理RabbitListener消费逻辑的线程决定了有没有限制住线程数量。如有2个队列监听，就有最多40个线程；有10个队列监听就有最多200个线程。

这里将通用消息消费着并发数减小，具体消费时用@RabbitListener去重置并发消费者数,再或者增加接口，按队列可动态调整并发消费者。

    SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();factory.setConnectionFactory(connectionFactory);factory.setPrefetchCount(5); // 每个消费者每次监听可拉取消息的数量factory.setMaxConcurrentConsumers(4); // 并发消费者最大值factory.setConcurrentConsumers(2); // 并发消费者初始值factory.setDefaultRequeueRejected(false); // 消息处理异常时是否丢弃消息，设置true有死循环风险factory.setAcknowledgeMode(AcknowledgeMode.AUTO); // 自动应答模式，消息接收处理异常情况下丢失消息return factory;@RabbitListener(bindings =@QueueBinding(value =@Queue(value = QUEUE,ignoreDeclarationExceptions = "true"),exchange =@Exchange(value = EVENT_EXCHANGE,ignoreDeclarationExceptions = "true",type = ExchangeTypes.TOPIC),key = ROUTING_KEY),concurrency = "4-8")