责任链模式+CompletableFuture异步处理

1、查询商品基础信息
2、查询商品价格
3、查询商品活动
4、查询商品库存
假设这几个服务逻辑比较独立，其实是可以并行调用，我们可以结合责任链模式和CompletableFuture进行优化:

下面是代码示例:

@Service
public class ChainFactory {
    // 原型模式获取对象

    @Resource
    private  Map<String, ILogicChain> strategyChainGroup;

    public ChainFactory(Map<String, ILogicChain> strategyChainGroup) {
        this.strategyChainGroup = strategyChainGroup;
    }

    /**
     * 构建责任链并获取所有处理器的 CompletableFuture 数组
     *
     * @param productInfo 商品信息对象
     * @return CompletableFuture 数组
     */
    public CompletableFuture<Void>[] openLogicChain(ProductInfo productInfo) {
        // 按照配置顺序装填责任链；
        ILogicChain logicChain = strategyChainGroup.get(LogicModel.BASIC_INFO_SERVICE_HANDLER.getCode());
        ILogicChain current = logicChain;
        List<CompletableFuture<Void>> futures = new ArrayList<>();

        //遍历LogicModel的code
        LogicModel[] ruleModels = LogicModel.values();
        for (int i = 1; i < ruleModels.length; i++) {
            ILogicChain nextChain = strategyChainGroup.get(ruleModels[i].getCode());
            current.appendNext(nextChain);
        }

        // 获取所有处理器的 CompletableFuture
        current = logicChain;
        while (current != null) {
            CompletableFuture<Void> future = current.logic(productInfo);
            futures.add(future);
            current = current.next();
        }

        return futures.toArray(new CompletableFuture[0]);
    }


    @Getter
    @AllArgsConstructor
    public enum LogicModel {

        BASIC_INFO_SERVICE_HANDLER("basicInfoServiceHandler", "实现商品基础信息补全节点"),
        PRICE_SERVICE_HANDLER("priceServiceHandler", "实现商品价格补全节点"),
        ACTIVITY_SERVICE_HANDLER("activityServiceHandler", "实现商品活动补全节点"),
        INVENTORY_SERVICE_HANDLER("inventoryServiceHandler", "实现商品库存服务节点")
        ;

        private final String code;
        private final String info;

    }

}

@Slf4j
@Component("activityServiceHandler")
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
public class ActivityServiceHandler extends AbstractLogicChain {
    @Override
    public CompletableFuture<Void> logic(ProductInfo productInfo) {
        CompletableFuture<Void> future = new CompletableFuture<>();
        // 模拟异步调用外部服务，修改商品信息对象
        CompletableFuture.runAsync(() -> {
            // 补充商品活动数据
            productInfo.setActivity("...");
            //休眠10秒
            try {
                TimeUnit.SECONDS.sleep(10);
                log.info("activityServiceHandler logic 结束:{}",now());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            future.complete(null);
        });
        return future;
    }

}

@Slf4j
@Component("basicInfoServiceHandler")
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
public class BasicInfoServiceHandler extends AbstractLogicChain {


    @Override
    public CompletableFuture<Void> logic(ProductInfo productInfo) {
        CompletableFuture<Void> future = new CompletableFuture<>();
        // 模拟异步调用外部服务，修改商品信息对象
        CompletableFuture.runAsync(() -> {
            // 补充商品基础信息数据
            productInfo.setBasicInfo("...");
            //休眠10秒
            try {
                TimeUnit.SECONDS.sleep(10);
                log.info("basicInfoServiceHandler logic 结束:{}",now());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            future.complete(null);
        });
        return future;
    }

}

@Slf4j
@Component("inventoryServiceHandler")
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
public class InventoryServiceHandler extends AbstractLogicChain {


    @Override
    public CompletableFuture<Void> logic(ProductInfo productInfo) {
        CompletableFuture<Void> future = new CompletableFuture<>();
        // 模拟异步调用外部服务，修改商品信息对象
        CompletableFuture.runAsync(() -> {
            // 补充商品库存数据
            productInfo.setInventory(100);
            //休眠10秒
            try {
                TimeUnit.SECONDS.sleep(10);
                log.info("inventoryServiceHandler logic 结束:{}",now());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            future.complete(null);
        });
        return future;
    }
}

@Slf4j
@Component("priceServiceHandler")
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
public class PriceServiceHandler extends AbstractLogicChain {


    @Override
    public CompletableFuture<Void> logic(ProductInfo productInfo) {
        CompletableFuture<Void> future = new CompletableFuture<>();
        // 模拟异步调用外部服务，修改商品信息对象
        CompletableFuture.runAsync(() -> {
            // 补充商品价格数据
            productInfo.setPrice(100.0);
            //休眠10秒
            try {
                TimeUnit.SECONDS.sleep(10);
                log.info("priceServiceHandler logic 结束:{}",now());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            future.complete(null);
        });
        return future;
    }


}

@Data
public class ProductInfo {
    //模拟商品基础信息
    private String basicInfo;
    //商品价格
    private double price;
    //模拟商品活动信息
    private String activity;
    //模拟商品库存
    private int inventory;

    // 省略构造函数、getter和setter
}

@Slf4j
public abstract class AbstractLogicChain implements ILogicChain {

    private ILogicChain next;

    @Override
    public ILogicChain next() {
        return next;
    }

    @Override
    public ILogicChain appendNext(ILogicChain next) {
        this.next = next;
        return next;
    }


}

public interface ILogicChain extends ILogicChainArmory, Cloneable {

    /**
     * 责任链接口
     *
     * @param productInfo    请求对象
     * @return
     */
    CompletableFuture<Void> logic(ProductInfo productInfo);

}

public interface ILogicChainArmory {

    ILogicChain next();

    ILogicChain appendNext(ILogicChain next);

}

测试类

@Resource
    private ChainFactory chainFactory;


    @Test
    public void testAsyncChain() {
        // 创建一个 ProductInfo 对象
        ProductInfo productInfo = new ProductInfo();

// 获取所有处理器的 CompletableFuture 数组
        CompletableFuture<Void>[] futures = chainFactory.openLogicChain(productInfo);

        // 等待所有责任链处理完成
        CompletableFuture.allOf(futures).join();

        log.info("结束");
    }

原文地址：https://blog.csdn.net/weixin_43076660/article/details/140371525

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！