性能优化之Hystrix请求合并&自实现简化版本

背景介绍

在业务开发过程中,存在这样的场景:程序接收到数据后,调用其他接口再将数据转发出去;如果接收一条转发一条,效率是比较低的,所以一个思路是先将数据缓存起来,缓存到一定数量后一次性转发出去。

10余年的柳江网站建设经验,针对设计、前端、开发、售后、文案、推广等六对一服务,响应快,48小时及时工作处理。营销型网站的优势是能够根据用户设备显示端的尺寸不同,自动调整柳江建站的显示方式,使网站能够适用不同显示终端,在浏览器中调整网站的宽度,无论在任何一种浏览器上浏览网站,都能展现优雅布局与设计,从而大程度地提升浏览体验。成都创新互联从事“柳江网站设计”,“柳江网站推广”以来,每个客户项目都认真落实执行。

有优点就有缺点,需要根据业务场景进行考量:

  • 在QPS较小的情况下,达到阈值的等待时间较长,造成数据延迟较大
  • 在应用发布的时候,缓存的数据存在丢失的可能性
  • 在应用非正常down掉的情况下,缓存的数据存在丢失的可能性

下面内容是对Hystrix请求合并及根据Hystrix请求合并原理自定义实现的简化版本。

Hystrix请求合并

什么是请求合并

Without Collapsing

without collapsing

With Collapsing

with collapsing

请求合并设计思路

design

Hystrix使用示例

示例采用Spring-Boot编写,下面代码拷贝到工程中可以直接运行。

添加依赖

下面是spring与hystrix集成的依赖pom。

pom


org.springframework.cloud
spring-cloud-starter-netflix-hystrix
2.1.6.RELEASE

启动类

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.netflix.hystrix.EnableHystrix;

@EnableHystrix
@SpringBootApplication
public class Application {
public static void main(String[] args){
SpringApplication.run(Application.class, args);
}
}

使用示例

HystrixController

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

import javax.annotation.Resource;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

@RestController
public class HystrixController {
@Resource
private HystrixService hystrixService;

@RequestMapping("/byid")
public Long byId(Long id) throws InterruptedException, ExecutionException {
Future future = hystrixService.byId(id);
return future.get();
}
}

HystrixService

import org.springframework.stereotype.Service;

import java.util.List;
import java.util.concurrent.Future;
import com.netflix.hystrix.contrib.javanica.annotation.HystrixCollapser;
import com.netflix.hystrix.contrib.javanica.annotation.HystrixCommand;
import com.netflix.hystrix.contrib.javanica.annotation.HystrixProperty;

@Service
public class HystrixService {
@HystrixCollapser(batchMethod="byIds",scope= com.netflix.hystrix.HystrixCollapser.Scope.GLOBAL,
collapserProperties={
@HystrixProperty(name="maxRequestsInBatch",value="10"),
@HystrixProperty(name="timerDelayInMilliseconds",value="1000")
})
public Future byId(Long id){
return null;
}

@HystrixCommand
public List byIds(List ids){
System.out.println(ids);
return ids;
}
}

测试类

发送请求进行验证。

import org.apache.http.HttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class HystrixTest {

public static void main(String[] args) throws Exception{
CloseableHttpClient httpClient = HttpClients.custom().setMaxConnPerRoute(100).build();
String url = "http://localhost:8086/byid?id=";
ExecutorService executorService = Executors.newFixedThreadPool(20);

int requestCount = 20;
for(int i = 0;i < requestCount;i++){
final int id = i;
executorService.execute(new Runnable() {
@Override
public void run(){
try{
HttpGet httpGet = new HttpGet(url+ id);
HttpResponse response = httpClient.execute(httpGet);
System.out.println(response);
}catch (Exception e){
e.printStackTrace();
}
}
});
}

Thread.sleep(1000*10);
executorService.shutdown();
httpClient.close();
}
}

简化版本实现

由于Hystrix已不再维护,同时考虑到Hystrix使用RxJava的学习门槛,根据HystrixCollapser设计思路及常见业务功能需求实现了一个简化版本。

实现

RequestCollapserFactory

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public class RequestCollapserFactory {
private static RequestCollapserFactory factory;
private static final int MAXBATCHSIZE = 20;
private static final long DELAY = 1000L;
private static final long PERIOD = 500L;
private ConcurrentHashMap collapsers;
private ScheduledExecutorService executor;

private RequestCollapserFactory(){
collapsers = new ConcurrentHashMap<>();
ThreadFactory threadFactory = new ThreadFactory() {
final AtomicInteger counter = new AtomicInteger();
@Override
public Thread newThread(Runnable r){
Thread thread = new Thread(r, "RequestCollapserTimer-" + counter.incrementAndGet());
thread.setDaemon(true);
return thread;
}
};
executor = new ScheduledThreadPoolExecutor(Runtime.getRuntime().availableProcessors(), threadFactory);
}

public static RequestCollapserFactory getInstance(){
if(factory != null){
return factory;
}
synchronized (RequestCollapserFactory.class){
if(factory != null){
return factory;
}
factory = new RequestCollapserFactory();
}
return factory;
}

public RequestCollapser getRequestCollapser(String key,RequestBatch requestBatch){
return getRequestCollapser(key,requestBatch,MAXBATCHSIZE,DELAY,PERIOD);
}
public RequestCollapser getRequestCollapser(String key,RequestBatch requestBatch,int maxBatchSize){
return getRequestCollapser(key,requestBatch,maxBatchSize,DELAY,PERIOD);
}
public RequestCollapser getRequestCollapser(String key,RequestBatch requestBatch,int maxBatchSize,long delay, long period){
RequestCollapser collapser = collapsers.get(key);
if(collapser != null){
return collapser;
}

synchronized (collapsers){
collapser = collapsers.get(key);
if(collapser != null){
return collapser;
}
collapser = new RequestCollapser(requestBatch,maxBatchSize,delay,period,executor);
collapsers.put(key,collapser);
return collapser;
}
}
}

RequestCollapser

import java.util.Collection;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

public class RequestCollapser {
private int maxBatchSize;
private long delay;
private long period;
private LinkedBlockingQueue queue = new LinkedBlockingQueue();
private ScheduledExecutorService executor;
private RequestBatch requestBatch;
private AtomicBoolean timerscheduled = new AtomicBoolean();
protected RequestCollapser(RequestBatch requestBatch,int maxBatchSize,long delay,long period,ScheduledExecutorService executor){
if(requestBatch == null){
throw new IllegalArgumentException("requestBatch can not be null");
}
this.maxBatchSize = maxBatchSize;
this.delay = delay;
this.period = period;
this.executor = executor;
this.requestBatch = requestBatch;
}

public List submitRequest(Object obj,boolean ifFullThenBatchExecute){
if(timerscheduled.compareAndSet(false,true)){
this.startSchedule();
}
List objectList = null;
synchronized (queue){
if(obj instanceof Collection){
queue.addAll((Collection)obj);
}else {
queue.offer(obj);
}

if(queue.size() >= this.maxBatchSize){
objectList = new LinkedList<>();
queue.drainTo(objectList);
}
}

if(!ifFullThenBatchExecute){
return objectList;
}

this.doBatch(objectList);
return null;
}

private boolean doBatch(List objectList){
if(objectList == null){
return true;
}
try{
return requestBatch.batch(objectList);
}catch (Throwable t){
t.printStackTrace();
}
return false;
}

private void startSchedule(){
Runnable r = new Runnable() {
@Override
public void run(){
List objectList = null;
synchronized (queue){
if(queue.size() > 0) {
objectList = new LinkedList<>();
queue.drainTo(objectList);
}
}
doBatch(objectList);
}
};

this.executor.scheduleAtFixedRate(r,this.delay,this.period, TimeUnit.MILLISECONDS);
}
}

RequestBatch

import java.util.List;

public interface RequestBatch {
boolean batch(List objectList);
}

验证测试

RequestCollapserTest

import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

public class RequestCollapserTest {
private static AtomicInteger counter = new AtomicInteger();
private static long delay = 1;
private static long period = 1;
private static int maxBatchSize = 20;
private static int requestCount = 50000;
private static RequestCollapserFactory factory = RequestCollapserFactory.getInstance();
private static RequestBatch requestBatch = new RequestBatch() {
@Override
public boolean batch(List objectList){
int size = objectList.size();
counter.addAndGet(size);
System.out.println(counter + ":::::" + size + ":::::" + objectList);
return true;
}
};

public static void main(String[] args) throws Exception{
//sync();
async();
}

public static void async() throws Exception{
ExecutorService executorService = Executors.newFixedThreadPool(20);
CountDownLatch countDownLatch = new CountDownLatch(requestCount);
for(int i = 0;i < requestCount;i++){
final int id = i;
executorService.execute(new Runnable() {
@Override
public void run(){
try{
RequestCollapser requestCollapser =
factory.getRequestCollapser("1",requestBatch,maxBatchSize,delay,period);
requestCollapser.submitRequest(id,true);
}catch (Exception e){
e.printStackTrace();
}
countDownLatch.countDown();
}
});
}

executorService.shutdown();
countDownLatch.await();
Thread.sleep(1000);
System.out.println(counter.get());
}

public static void sync() throws Exception{
for(int i = 0;i < requestCount;i++){
final int id = i;
RequestCollapser requestCollapser =
factory.getRequestCollapser("1",requestBatch,maxBatchSize,delay,period);
requestCollapser.submitRequest(id,true);
}
Thread.sleep(1000);
System.out.println(counter.get());
}
}

网站名称:性能优化之Hystrix请求合并&自实现简化版本
网站路径:http://www.gawzjz.com/qtweb2/news34/9184.html

网站建设、网络推广公司-创新互联,是专注品牌与效果的网站制作,网络营销seo公司;服务项目有等

广告

声明:本网站发布的内容(图片、视频和文字)以用户投稿、用户转载内容为主,如果涉及侵权请尽快告知,我们将会在第一时间删除。文章观点不代表本网站立场,如需处理请联系客服。电话:028-86922220;邮箱:631063699@qq.com。内容未经允许不得转载,或转载时需注明来源: 创新互联