hystrix 实现请求合并

需要请求合并的服务启动类上添加

@EnableHystrix
@EnableHystrixDashboard

需要请求合并的服务添加依赖

<!--hystrix依赖-->
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-netflix-hystrix</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-netflix-hystrix-dashboard</artifactId>
</dependency>

请求接口

package com.leilei.controller;

import com.leilei.entity.AlarmDictionary;
import com.leilei.service.AlarmService;
import com.netflix.hystrix.strategy.concurrency.HystrixRequestContext;
import org.springframework.web.bind.annotation.*;

import java.util.concurrent.Future;

/** * @author lei * @create 2022-03-30 22:45 * @desc **/
@RequestMapping("/alarm")
@RestController
public class AlarmController {
    
    private final AlarmService alarmService;

    public AlarmController(AlarmService alarmService) {
    
        this.alarmService = alarmService;
    }

    /** * 外部单个请求接口 * @param id * @return */
    @GetMapping(value = "/dictionary/{id}")
    public AlarmDictionary requestCollapseAnnotation(@PathVariable Integer id) {
    
        // 这里注意需要开启Hystrix环境
        try (HystrixRequestContext context = HystrixRequestContext.initializeContext()) {
    
            Future<AlarmDictionary> f1 = alarmService.find(id);
            return f1.get();
        } catch (Exception e) {
    
            e.printStackTrace();
        }
        return null;
    }
}

服务内部进行聚合批量处理

package com.leilei.service;

import com.leilei.entity.AlarmDictionary;
import com.netflix.hystrix.contrib.javanica.annotation.HystrixCollapser;
import com.netflix.hystrix.contrib.javanica.annotation.HystrixCommand;
import com.netflix.hystrix.contrib.javanica.annotation.HystrixProperty;
import org.springframework.stereotype.Service;

import java.util.List;
import java.util.concurrent.Future;

/** * @author lei * @create 2022-03-30 22:38 * @desc **/
@Service
public class AlarmService {
    

    private  final AlarmFeign alarmFeign;

    public AlarmService(AlarmFeign alarmFeign) {
    
        this.alarmFeign = alarmFeign;
    }

    /** * batchMethod 设置批量方法的接口名 * HystrixProperty name 设置为时间延迟timerDelayInMilliseconds 即多少时间的请求作为一个批量进行提交；value 为时间阈值，单位是毫秒 * @param id * @return */
    @HystrixCollapser(batchMethod = "findBatch",
            collapserProperties = {
    @HystrixProperty(name = "timerDelayInMilliseconds", value = "1000")})
    public Future<AlarmDictionary> find(Integer id) {
    
        throw new RuntimeException("This method body should not be executed");
    }

    @HystrixCommand
    public List<AlarmDictionary> findBatch(List<Integer> alarms) {
    
        List<AlarmDictionary> alarmDictionaries = alarmFeign.getBatch(alarms);
        System.out.println(Thread.currentThread().getName() + ":" + alarmDictionaries);
        return alarmDictionaries;
    }
}

服务提供者(当然这里也不可以用远程，使用服务内部批量逻辑一样)

@RestController
@RequestMapping("/alarm/dictionary")
public class AlarmDictionaryController {
    static List<AlarmDictionary> alarms = new ArrayList<>();
    static {
        alarms.add(new AlarmDictionary(1,"808一级报警"));
        alarms.add(new AlarmDictionary(2,"808二级报警"));
        alarms.add(new AlarmDictionary(3,"808三级报警"));
        alarms.add(new AlarmDictionary(4,"疲劳驾驶一级报警"));
    }
    @PostMapping("/batch")
    public List<AlarmDictionary> getBatchAlarmById(@RequestBody List<Integer> alarmIds) {
        return alarms.stream().filter(x -> alarmIds.contains(x.getId())).collect(Collectors.toList());
    }
}

测试结果

问题:

Hystrix请求必须做到请求应答模式，就是说，一个请求必须要有对应的结果，如果响应为空，则会抛出异常

Caused by: java.lang.RuntimeException: Failed to map all collapsed requests to response. The expected contract has not been respected. Collapser key: 'find', requests size: '1', response size: '0'
	at com.netflix.hystrix.contrib.javanica.collapser.CommandCollapser.mapResponseToRequests(CommandCollapser.java:76)
	at com.netflix.hystrix.contrib.javanica.collapser.CommandCollapser.mapResponseToRequests(CommandCollapser.java:33)
	at com.netflix.hystrix.HystrixCollapser$1$1.call(HystrixCollapser.java:171)
	at rx.internal.util.ActionObserver.onNext(ActionObserver.java:39)
	at rx.internal.operators.OnSubscribeDoOnEach$DoOnEachSubscriber.onNext(OnSubscribeDoOnEach.java:96)
	... 68 more

当前我们可以对其进行try-catch处理，更好的解决方式待更新…

TODO…

附上项目地址:hystrix-batch-collapser