Merge branch 'api_score_meger'

935090232@qq.com

2022-02-20 c253b555c7905c5136d47cd615ef545fa50cc6ad

 zq-erp/src/main/java/com/matrix/component/asyncmessage/AsyncMessageManager.java

New file
@@ -0,0 +1,117 @@
/**
 * projectName: zq-erp
 * fileName: MessageManager.java
 * packageName: com.matrix.component.asyncmessage
 * date: 2021-10-18 14:01
 * copyright(c) 2021 http://www.hydee.cn/ Inc. All rights reserved.
 */
package com.matrix.component.asyncmessage;

import cn.hutool.core.collection.CollectionUtil;
import com.matrix.core.tools.LogUtil;
import com.matrix.core.tools.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.core.Ordered;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;

import java.util.*;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.locks.ReentrantLock;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

/**
 * @version: V1.0
 * @author: JiangYouYao
 * @className: AsyncMessageManager
 * @packageName: com.matrix.component.asyncmessage
 * @description: 异步消息管理者
 * @data: 2021-10-18 14:01
 **/
@Component
@Order(Ordered.HIGHEST_PRECEDENCE)
public class AsyncMessageManager implements ApplicationRunner {

    @Autowired
    private List<MessageHandler> obs;

    private Map<String, List<MessageHandler>> routes;


    @Override
    public void run(ApplicationArguments args) throws Exception {
        if (CollectionUtil.isNotEmpty(obs)) {
            routes = obs.stream().collect(Collectors.groupingBy(MessageHandler::getRouteKey));
            LogUtil.info("异步消息绑定成功，检测到{} 个消费者，共计{}组", obs.size(), routes.size());
        } else {
            LogUtil.info("未检测到异步消息处理类");
        }
    }




    /**
     * map 参数的字符串表示，方便快速拼装消息参数
     * @param routeKey
     * @param mapStr
     */
    public void sendMsg(String routeKey, String mapStr,Object... args){

        if(StringUtils.isBlank(mapStr)){
            throw new IllegalArgumentException("mapStr格式错误：例如：\\\"orderId=123,price=88\\\"，mapStr="+mapStr);
        }
        mapStr=String.format(mapStr,args);

        Map<String, Object> param =new HashMap<>();
        String[] paramStr = mapStr.split(",");
        Arrays.asList(paramStr).forEach(item->{
            String[] keyValueArr = item.split("=");
            param.put(keyValueArr[0],keyValueArr[1]);
        });
        sendMsg(routeKey,param);
    }

    /**
     * 根据route 发送消息到对应的消费者
     * 这个方法本质上还是同步执行，没有完成效率上的异步解耦，这里做观察者模式主要是为了扩展业务，减少对第三方消息组件的依赖
     * 而不是解决性能问题，如果后续需要解决性能问题，在加一个消息队列，然后启动线程从队列中进行消息的消费。
     * @param routeKey
     * @param param
     */
    public void sendMsg(String routeKey, Map<String, Object> param) {

        if(StringUtils.isBlank(routeKey)){
            LogUtil.warn("发送异步消息失败：routeKey为空");
            return;
        }

        //匹配观察者
        List<MessageHandler> lisener = new ArrayList<>();
        for (Map.Entry<String, List<MessageHandler>> routesEntry : routes.entrySet()) {
            if (Pattern.matches(routesEntry.getKey(), routeKey)) {
                lisener.addAll(routesEntry.getValue());
            }
        }

        //通知观察者
        if (CollectionUtil.isNotEmpty(lisener)) {
            LogUtil.info("发送异步消息，routeKey={}，匹配观察者{}个",routeKey,lisener.size());
            for (MessageHandler messageHandler : lisener) {
                try{
                    messageHandler.handle(param);
                }catch (Throwable t){
                    LogUtil.error("{}，处理类执行异常:routeKey={},message={}", messageHandler.getName(),routeKey,t.getMessage());
                }
            }
        }else{
            LogUtil.warn("未匹配到routeKey={}，的消费者",routeKey);
        }

    }


}