基于springcloud异步线程池、高并发请求feign的解决方案

基于springcloud异步线程池、高并发请求feign的解决方案

ScenTaskTestApplication.java

package com.test;

import org.springframework.boot.SpringApplication;

import org.springframework.boot.autoconfigure.SpringBootApplication;

import org.springframework.cloud.openfeign.EnableFeignClients;

/**

* @author scen

* @version 2018年9月27日 上午11:51:04

*/

@EnableFeignClients

@SpringBootApplication

public class ScenTaskTestApplication {

public static void main(String[] args) {

SpringApplication.run(ScenTaskTestApplication.class, args);

}

}

application.properties

spring.application.name=scen-task-test

server.port=9009

feign.hystrix.enabled=true

#熔断器失败的个数==进入熔断器的请求达到1000时服务降级（之后的请求直接进入熔断器）

hystrix.command.default.circuitBreaker.requestVolumeThreshold=1000

#回退最大线程数

hystrix.command.default.fallback.isolation.semaphore.maxConcurrentRequests=50

#核心线程池数量

hystrix.threadpool.default.coreSize=130

#请求处理的超时时间

hystrix.command.default.execution.isolation.thread.timeoutInMilliseconds=100000

ribbon.ReadTimeout=120000

#请求连接的超时时间

ribbon.ConnectTimeout=130000

eureka.instance.instance-id=${spring.application.name}:${spring.application.instance_id:${server.port}}

eureka.instance.preferIpAddress=true

eureka.client.service-url.defaultZone=http://127.0.0.1:9000/eureka

logging.level.com.test.user.service=debug

logging.level.org.springframework.boot=debug

logging.level.custom=info

AsyncConfig.java

package com.test;

import org.springframework.context.annotation.Configuration;

import org.springframework.scheduling.annotation.AsyncConfigurer;

import org.springframework.scheduling.annotation.EnableAsync;

import org.springframework.scheduling.concurrent.ThreadPoolTahttp://skExecutor;

import java.util.concurrent.Executor;

/**

* springboot异步线程池配置

* @author Scen

* @date 2018/11/7 18:28

*/

@Configuration

@EnableAsync

public class AsyncConfig implements AsyncConfigurer {

@Override

public Executor getAsyncExecutor() {

//定义线程池

ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();

//核心线程数

taskExecutor.setCorePoolSize(20);

//线程池最大线程数

taskExecutor.setMaxPoolSize(100);

//线程队列最大线程数

taskExecutor.setQueueCapacity(10);

//初始化

taskExecutor.initialize();

return taskExecutor;

}

}

DoTaskClass.java

package com.test;

import com.test.pojo.User;

import com.test.pojo.UserEducation;

import com.test.user.service.UserService;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.scheduling.annotation.Async;

import org.springframework.stereotype.Component;

import java.util.List;

/**

* 任务类 定义异步工作任务

* @author Scen

* @date 2018/11/7 18:40

*/

@Component

public class DoTaskClass {

/**

* 一个feign的客户端

*/

private final UserService userService;

@Autowired

public DoTaskClass(UserService userService) {

this.userServmOdlSdmLice = userService;

}

/**

* 核心任务

*

* @param uid

*/

@Async

public void dotask(String uid) {

/**

* 模拟复杂工作业务（109个线程同时通过feign请求微服务提供者）

*/

{

List userEducationByUid = userService.findUserEducationByUid(uid);

List blackList = userService.getBlackList();

String userSkilled = userService.getUserSkilled(uid);

String userFollow = userService.getUserFollow(uid);

User userById = userService.getUserById(uid);

List followList = userService.getFollowList(uid);

int userActivityScore = userService.getUserActivityScore(uid);

}

// 打印线程名称分辨是否为多线程操作

System.out.println(Thread.currentThread().getName() + "===任务" + uid + "执行完成===");

}

}

TestController.java

package com.test;

import com.test.pojo.User;

import com.test.user.service.UserService;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.web.bind.annotation.RequestMapping;

import org.springframework.web.bind.annotation.RestController;

import java.util.List;

/**

* 测试案例

* @author Scen

* @date 2018/11/7 18:10

*/

@RestController

public class TestController {

/**

* 此处仅用此feign客户端请求微服务获取核心工作所需参数

*/

private final UserService userService;

/**

* 核心工作异步算法

*/

private final DoTaskClass doTaskClass;

@Autowired

public Teshttp://tController(DoTaskClass doTaskClass, UserService userService) {

this.doTaskClass = doTaskClass;

this.userService = userService;

}

/**

* 手动触发工作

* @throws InterruptedException

*/

@RequestMapping("/test")

public void task() throws InterruptedException {

/*

取到1000个要执行任务的必备参数

*/

List userList = userService.findAllLite(1, 1000);

for (int i = 0; i < userList.size(); i++) {

try {

// 异步线程开始工作http://

doTaskClass.dotask(userList.get(i).getId());

} catch (Exception e) {

/*

若并发线程数达到MaxPoolSize+QueueCapacity=110（参考AsyncConfig配置）会进入catch代码块

i--休眠3秒后重试（尝试进入线程队列：当且仅当109个线程有一个或多个线程完成异步任务时重试成功）

*/

i--;

Thread.sleep(3000*3);

}

System.out.println(i);

}

}

}

相关线程池、超时时间等数量和大小按实际业务配置

补充：SpringCloud关于@FeignClient和Hystrix集成对http线程池监控问题

@FeignClient可以作为Http代理访问其他微服务节点，可以用apache的httpclient替换@FeignClient原生的URLConnection请求方式，以达到让http请求走Http线程池的目的。

而@FeignClient和hystrix集成之后，在hystrix dashboard上可以监控到 @FeignClient 中接口调用情况和 @FeignClient 中httpclient中线程池使用状况。

下面是demo的示例：

1、@FeignClient的接口代码如下：

@FeignClient(value="service-A", fallback=ServiceClientHystrix.class)

public interface ServiceClient {

@RequestMapping(method = RequestMethod.GET, value = "/add/{id}")

String add(@PathVariable("id") Integer id);

}

2、ServiceClientHystrix.java

@Component

public class ServiceClientHystrix implements ServiceClient{

@Override

public String add(Integer id) {

return "add value from ServiceClientHystrix";

}

}

3、关于@FeignClient和hystrix

集成后，Http线程池配置如下：

hystrix.threadpool.服务实例ID.参数

例如设置httpclient的线程池最大线程数量

hystrix.threadpool.service-A.coreSize=20//默认是hystrix.threadpool.default.coreSize = 10

hystrix.threadpool.service-A.maximumSize=20//默认是hystrix.threadpool.default.maximumSize = 10

启动服务后用测试用例连续调用接口测试，用hystrix dashboard

监控得到下图监控效果：

去掉hystrix.threadpool.服务实例ID.参数配置后，再次用测试用例调用接口得到监控如下图：

PoolSize的大小取决于hystrix.threadpool.服务实例ID.coreSize大小设置

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。