concurrent-aggregatorSpring 無(wú)關(guān)聯(lián)并發(fā)聚合器

背景與目的

在開(kāi)發(fā)后臺(tái)接口時(shí), 為了開(kāi)發(fā)效率, 我們往往習(xí)慣編寫(xiě)串行執(zhí)行的代碼, 去調(diào)用不同的接口,

即使這些接口之間并無(wú)依賴, 這使得最后開(kāi)發(fā)的接口性能低下, 且數(shù)據(jù)不方便復(fù)用

此框架目的旨在保持開(kāi)發(fā)效率的同時(shí), 很方便地支持并發(fā)和數(shù)據(jù)復(fù)用

原理

1. Spring + CountDownLatch + Future + 反射 + 動(dòng)態(tài)代理
2. 通過(guò)啟動(dòng)類(lèi)的注解去加載需要代理的接口，
3. 用 Factorybean 將代理的bean注冊(cè)到spring 容器
4. 代理接口使用注解配置使用 spring 的 bean 來(lái)完成操作
5. 線程池執(zhí)行具體的操作，CountDownLatch做同步
6. 目前自動(dòng)識(shí)別代理方法的類(lèi)型和參數(shù)比較弱雞，后續(xù)需要繼續(xù)完善
7. 工程需要自己下載下來(lái) install

注解介紹

• @EnableConcurrentAggregate 定義數(shù)據(jù)提供者
• @ConcurrentAggregate 動(dòng)態(tài)代理的接口上需要添加注解
• @AggregateProviders 并行處理的提供者注解
• @AggregateProvider  定單個(gè)并行處理的方法的注解
• @AggregateConsumer   最后聚合的方注解
• @AggregateParameterName 參數(shù)名

使用方法

pom.xml

<dependency>
    <groupId>com.wolfking</groupId>
    <artifactId>concurrent-aggregator</artifactId>
    <version>1.0-SNAPSHOT</version>
</dependency>

啟動(dòng)類(lèi)添加注解

    @EnableConcurrentAggregate("com.xxx.service.xxx.aggregator")

示例

開(kāi)發(fā)h5首頁(yè)接口，需要從七八個(gè)地方組織數(shù)據(jù)，并且數(shù)據(jù)沒(méi)有關(guān)聯(lián)，7個(gè)接口串聯(lián)調(diào)用需要3秒

IndexService.java
IndexService.java
    1. IndexVo assemblyExperience(IndexVo indexVo, Integer tenantId)
    2. IndexVo assemblyTheme(IndexVo indexVo, Integer tenantId)
    3. IndexVo assemblyFounder(IndexVo indexVo, Integer tenantId, Integer customerId)
    4. IndexVo assemblyRaffleActivity(IndexVo indexVo, Integer tenantId)
    5. IndexVo assemblyAgreeActivity(IndexVo indexVo, Integer tenantId, Integer customerId, Integer userId)
    6. IndexVo assemblyBargin(IndexVo indexVo, Integer tenantId, Integer customerId)
    7. IndexVo assemblyGroup(IndexVo indexVo, Integer tenantId, Integer customerId)
    8. IndexVo assemblyGroup(IndexVo indexVo, Integer tenantId, Integer customerId)
    9. IndexVo assemblyReduction(IndexVo indexVo, Integer tenantId, Integer customerId)
    10.void summary()    ---匯聚接口

定義并發(fā)代理的接口

最后匯聚AggregateConsumer輸入?yún)?shù)是AggregateProvider每個(gè)的返回值，discardResult除外，匯聚的可以沒(méi)有

@ConcurrentAggregate
public interface IndexAggregatorProxy {

    @AggregateProviders(ignoreException = true, timeout = 5000, value = {
            @AggregateProvider(beanClass = IndexService.class, method = "assemblyExperience",
                    parameters = {@AggregateParameter("index"), @AggregateParameter("tenantId")}),
            @AggregateProvider(beanClass = IndexService.class, method = "assemblyTheme",
                    parameters = {@AggregateParameter("index"), @AggregateParameter("tenantId")}),
            @AggregateProvider(beanClass = IndexService.class, method = "assemblyFounder",
                    parameters = {@AggregateParameter("index"), @AggregateParameter("tenantId"), @AggregateParameter("customerId")}),
            @AggregateProvider(beanClass = IndexService.class, method = "assemblyRaffleActivity",
                    parameters = {@AggregateParameter("index"), @AggregateParameter("tenantId")}),
            @AggregateProvider(beanClass = IndexService.class, method = "assemblyAgreeActivity",
                    parameters = {@AggregateParameter("index"), @AggregateParameter("tenantId"), @AggregateParameter("customerId"), @AggregateParameter("userId")}),
            @AggregateProvider(beanClass = IndexService.class, method = "assemblyBargin",
                    parameters = {@AggregateParameter("index"), @AggregateParameter("tenantId"), @AggregateParameter("customerId")}),
            @AggregateProvider(beanClass = IndexService.class, method = "assemblyGroup",
                    parameters = {@AggregateParameter("index"), @AggregateParameter("tenantId"), @AggregateParameter("customerId")}),
            @AggregateProvider(beanClass = IndexService.class, method = "assemblyInviteFriend",
                    parameters = {@AggregateParameter("index"), @AggregateParameter("tenantId")}),
            @AggregateProvider(beanClass = IndexService.class, method = "assemblyReduction",
                    parameters = {@AggregateParameter("index"), @AggregateParameter("tenantId"), @AggregateParameter("customerId")})})
    @AggregateConsumer(beanClass = IndexService.class,method = "summary")
    void index(@AggregateParameterName("index") IndexVo indexVo,
               @AggregateParameterName("tenantId") Integer tenantId,
               @AggregateParameterName("customerId") Integer customerId,
               @AggregateParameterName("userId") Integer userId);
}

在spring的service中使用并發(fā)代理的接口，直接注入就行

    @Autowired
    @SuppressWarnings("all")
    private IndexAggregatorProxy indexAggregatorProxy;