• 成员变量为动态数据时不可轻易使用

    问题描述

    业务验收阶段，遇到了一个由于成员变量导致的线程问题

    有一个kafka切面，用来处理某些功能在调用前后的发送消息，资产类型type是成员变量定义；

    资产1类型推送消息是以zichan1为节点；资产2类型推送消息是以zichan2为节点；

    当多个线程调用切面类时，由于切面类中使用成员变量且为动态数据时，此时会出现根据资产类型推送消息错误；例如资产1在调用功能时，切面类的type字段为zichan1；同时有资产2调用功能时，此时的切面类的type字段为zichan2;导致资产1在调用功能前推送的是zichan1,在调用功能后推送的是zichan2的消息标识。

    原因

    当多个线程同时调用时，成员变量则会只采用最后一次调用的值。

    下面简单描述下情景：

    kafkaAspect类

    1. import com.alibaba.fastjson.JSON;
    2. import com.example.demo.constant.StageCodeConstant;
    3. import com.example.demo.entity.KafkaSendMessageConstant;
    4. import com.example.demo.util.DateUtils;
    5. import lombok.extern.slf4j.Slf4j;
    6. import org.apache.commons.lang.ArrayUtils;
    7. import org.aspectj.lang.ProceedingJoinPoint;
    8. import org.aspectj.lang.annotation.Around;
    9. import org.aspectj.lang.annotation.Aspect;
    10. import org.aspectj.lang.reflect.MethodSignature;
    11. import org.slf4j.Logger;
    12. import org.slf4j.LoggerFactory;
    13. import org.springframework.stereotype.Component;
    14.  
    15. @Component
    16. @Aspect
    17. @Slf4j
    18. public class KafkaProcessAspect {
    19.  
    20.     private static final Logger log = LoggerFactory.getLogger(KafkaProcessAspect.class);
    21.  
    22.     private String assetType = "";
    23.     private String startTime = "";
    24.  
    25.     @Around("@annotation(kafkaProcess)")
    26.     public Object doAround(ProceedingJoinPoint joinPoint, KafkaProcess kafkaProcess) throws Throwable {
    27.         Object object = null;
    28.         assetType = (String) getParamValue(joinPoint, "assetType");
    29.         startTime = DateUtils.getTime();
    30.         object = joinPoint.proceed();
    31.         //推送消息
    32.         String stageCode = StageCodeConstant.STAGE_CODE.get(assetType).get(kafkaProcess.functionName());
    33.         KafkaSendMessageConstant messageConstant = new KafkaSendMessageConstant();
    34.         messageConstant.setStageCode(stageCode);
    35.         messageConstant.setStartTime(startTime);
    36.         messageConstant.setEndTime(DateUtils.getTime());
    37.         log.info("资产类型{}推送消息：{}", assetType, JSON.toJSONString(messageConstant));
    38.         return object;
    39.     }
    40.  
    41.     private Object getParamValue(ProceedingJoinPoint joinPoint, String paramName) {
    42.         Object[] params = joinPoint.getArgs();
    43.         String[] parameterNames = ((MethodSignature) joinPoint.getSignature()).getParameterNames();
    44.  
    45.         if (parameterNames == null || parameterNames.length == 0) {
    46.             return null;
    47.         }
    48.         for (String param : parameterNames) {
    49.             if (param.equals(paramName)) {
    50.                 int index = ArrayUtils.indexOf(parameterNames, param);
    51.                 return params[index];
    52.             }
    53.         }
    54.         return null;
    55.     }
    56.  
    57. }

     由于controller中的请求地址是采用占位符定义，后使用@PathVariable可以获取的类型

    Controller类

    1. import com.example.demo.aop.KafkaProcess;
    2. import org.slf4j.Logger;
    3. import org.slf4j.LoggerFactory;
    4. import org.springframework.web.bind.annotation.PathVariable;
    5. import org.springframework.web.bind.annotation.PostMapping;
    6. import org.springframework.web.bind.annotation.RequestMapping;
    7. import org.springframework.web.bind.annotation.RestController;
    8.  
    9. @RestController
    10. @RequestMapping("/test")
    11. public class TestController {
    12.  
    13.     private static final Logger logger = LoggerFactory.getLogger(TestController.class);
    14.  
    15.     @PostMapping("/{assetType}/cmpt")
    16.     @KafkaProcess(functionName = "JS")
    17.     public void cmpt(@PathVariable("assetType") String assetType) {
    18.         logger.info("{}接口开始", assetType);
    19.         long startTime = System.currentTimeMillis();
    20.  
    21.         for (int i = 0; i < 20000; i++) {
    22.             for (int j = 0; j < 20000; j++) {
    23.                 int m = i * j;
    24.                 logger.debug("i*j={}",m);
    25.             }
    26.         }
    27.  
    28.         long endTime = System.currentTimeMillis();
    29.         logger.info("{}接口结束,耗时{}", assetType, endTime - startTime);
    30.     }
    31.  
    32. }

     资产类型枚举 AssetTypeEnum

    1. public enum AssetTypeEnum {
    2.  
    3.     ZICHAN_1("zichan1","资产1"),
    4.     ZICHAN_2("zichan2","资产2");
    5.  
    6.     // 成员变量
    7.     private String code;
    8.  
    9.     private String desc;
    10.  
    11.     public String getCode() {
    12.         return code;
    13.     }
    14.  
    15.     public void setCode(String code) {
    16.         this.code = code;
    17.     }
    18.  
    19.     public String getDesc() {
    20.         return desc;
    21.     }
    22.  
    23.     public void setDesc(String desc) {
    24.         this.desc = desc;
    25.     }
    26.  
    27.     // 构造方法
    28.     AssetTypeEnum(String code,String desc) {
    29.         this.code = code;
    30.         this.desc = desc;
    31.     }
    32.  
    33. }

    推送消息 KafkaSendMessageConstant

    1. public class KafkaSendMessageConstant {
    2.  
    3.     private String stageCode;
    4.     private String startTime;
    5.     private String endTime;
    6.  
    7.     public String getStageCode() {
    8.         return stageCode;
    9.     }
    10.  
    11.     public void setStageCode(String stageCode) {
    12.         this.stageCode = stageCode;
    13.     }
    14.  
    15.     public String getStartTime() {
    16.         return startTime;
    17.     }
    18.  
    19.     public void setStartTime(String startTime) {
    20.         this.startTime = startTime;
    21.     }
    22.  
    23.     public String getEndTime() {
    24.         return endTime;
    25.     }
    26.  
    27.     public void setEndTime(String endTime) {
    28.         this.endTime = endTime;
    29.     }
    30. }

    节点常量类 StageCodeConstant ；根据不同资产类型赋值不同功能的推送标识

    1. import java.util.HashMap;
    2. import java.util.Map;
    3.  
    4. public class StageCodeConstant {
    5.  
    6.     public static final Map> STAGE_CODE = new HashMap<>();
    7.  
    8.     static {
    9.         //资产1
    10.         STAGE_CODE.put(AssetTypeEnum.ZICHAN_1.getCode(),
    11.                 new HashMap() {{
    12.                     put("JS", "ZICHAN1-JS");
    13.                 }});
    14.         //资产2
    15.         STAGE_CODE.put(AssetTypeEnum.ZICHAN_2.getCode(),
    16.                 new HashMap() {{
    17.                     put("JS", "ZICHAN2-JS");
    18.                 }});
    19.     }
    20. }

    用postman调用资产2接口

    后立即调用资产1接口

    此时出现这种结果：

    会发现，调用资产2的时候发送消息还是资产1的信息；然后资产1发送的消息也是资产1的信息

    解决

    此时有两个解决办法，一个是将doAround()和其他方法合并为一个方法，将成员变量调整为局部变量；另一个则为将该成员变量设置为一个对象，对这个对象进行线程设置，保证doAround()和doBefore()获取的是同一个对象的数据

    解决方法一

    1. import com.alibaba.fastjson.JSON;
    2. import com.example.demo.constant.StageCodeConstant;
    3. import com.example.demo.entity.KafkaSendMessageConstant;
    4. import com.example.demo.util.DateUtils;
    5. import lombok.extern.slf4j.Slf4j;
    6. import org.apache.commons.lang.ArrayUtils;
    7. import org.aspectj.lang.ProceedingJoinPoint;
    8. import org.aspectj.lang.annotation.Around;
    9. import org.aspectj.lang.annotation.Aspect;
    10. import org.aspectj.lang.reflect.MethodSignature;
    11. import org.slf4j.Logger;
    12. import org.slf4j.LoggerFactory;
    13. import org.springframework.stereotype.Component;
    14.  
    15. @Component
    16. @Aspect
    17. @Slf4j
    18. public class KafkaProcessAspect {
    19.  
    20.     private static final Logger log = LoggerFactory.getLogger(KafkaProcessAspect.class);
    21.  
    22.     @Around("@annotation(kafkaProcess)")
    23.     public Object doAround(ProceedingJoinPoint joinPoint, KafkaProcess kafkaProcess) throws Throwable {
    24.         Object object = null;
    25.         String assetType = (String) getParamValue(joinPoint, "assetType");
    26.         String startTime = DateUtils.getTime();
    27.         object = joinPoint.proceed();
    28.         //推送消息
    29.         String stageCode = StageCodeConstant.STAGE_CODE.get(assetType).get(kafkaProcess.functionName());
    30.         KafkaSendMessageConstant messageConstant = new KafkaSendMessageConstant();
    31.         messageConstant.setStageCode(stageCode);
    32.         messageConstant.setStartTime(startTime);
    33.         messageConstant.setEndTime(DateUtils.getTime());
    34.         log.info("资产类型{}推送消息：{}", assetType, JSON.toJSONString(messageConstant));
    35.         return object;
    36.     }
    37.  
    38.     private Object getParamValue(ProceedingJoinPoint joinPoint, String paramName) {
    39.         Object[] params = joinPoint.getArgs();
    40.         String[] parameterNames = ((MethodSignature) joinPoint.getSignature()).getParameterNames();
    41.  
    42.         if (parameterNames == null || parameterNames.length == 0) {
    43.             return null;
    44.         }
    45.         for (String param : parameterNames) {
    46.             if (param.equals(paramName)) {
    47.                 int index = ArrayUtils.indexOf(parameterNames, param);
    48.                 return params[index];
    49.             }
    50.         }
    51.         return null;
    52.     }
    53.  
    54. }

    解决方法二

    成员变量KafkaSingleDTO

    1. public class KafkaSingleDTO {
    2.  
    3.     private String assetType="";
    4.     private String date="";
    5.  
    6.     public String getAssetType() {
    7.         return assetType;
    8.     }
    9.  
    10.     public void setAssetType(String assetType) {
    11.         this.assetType = assetType;
    12.     }
    13.  
    14.     public String getDate() {
    15.         return date;
    16.     }
    17.  
    18.     public void setDate(String date) {
    19.         this.date = date;
    20.     }
    21. }

    对象单实例获取 KafkaSingleUtil

    1. import com.example.demo.entity.KafkaSingleDTO;
    2.  
    3. public class KafkaSingleUtil {
    4.  
    5.     private static ThreadLocal START = new ThreadLocal<>();
    6.  
    7.     public static KafkaSingleDTO getObject() {
    8.         KafkaSingleDTO singleDTO = START.get();
    9.         if (singleDTO == null) {
    10.             singleDTO = new KafkaSingleDTO();
    11.         }
    12.         return singleDTO;
    13.     }
    14.  
    15. }

    kafkaAsspect拦截器

    1. import com.alibaba.fastjson.JSON;
    2. import com.example.demo.constant.StageCodeConstant;
    3. import com.example.demo.entity.KafkaSendMessageConstant;
    4. import com.example.demo.entity.KafkaSingleDTO;
    5. import com.example.demo.util.DateUtils;
    6. import lombok.extern.slf4j.Slf4j;
    7. import org.apache.commons.lang.ArrayUtils;
    8. import org.aspectj.lang.ProceedingJoinPoint;
    9. import org.aspectj.lang.annotation.Around;
    10. import org.aspectj.lang.annotation.Aspect;
    11. import org.aspectj.lang.annotation.Before;
    12. import org.aspectj.lang.reflect.MethodSignature;
    13. import org.slf4j.Logger;
    14. import org.slf4j.LoggerFactory;
    15. import org.springframework.stereotype.Component;
    16.  
    17. @Component
    18. @Aspect
    19. @Slf4j
    20. public class KafkaProcessAspect {
    21.  
    22.     private static final Logger log = LoggerFactory.getLogger(KafkaProcessAspect.class);
    23.  
    24.     private static String assetType="";
    25.     private static String startTime="";
    26.  
    27.     @Around("@annotation(kafkaProcess)")
    28.     public Object doAround(ProceedingJoinPoint joinPoint, KafkaProcess kafkaProcess) throws Throwable {
    29.         Object object = null;
    30.         assetType = (String) getParamValue(joinPoint, "assetType");
    31.         startTime = DateUtils.getTime();
    32.         KafkaSingleDTO singleDTO = KafkaSingleUtil.getObject();
    33.         singleDTO.setAssetType(assetType);
    34.         singleDTO.setDate(startTime);
    35.  
    36.         object = joinPoint.proceed();
    37.         //推送消息--方法调用后
    38.         String stageCode = StageCodeConstant.STAGE_CODE.get(singleDTO.getAssetType()).get(kafkaProcess.functionName());
    39.         KafkaSendMessageConstant messageConstant = new KafkaSendMessageConstant();
    40.         messageConstant.setStageCode(stageCode);
    41.         messageConstant.setStartTime(singleDTO.getDate());
    42.         messageConstant.setEndTime(DateUtils.getTime());
    43.         log.info("资产类型{}方法后推送消息：{}", singleDTO.getAssetType(), JSON.toJSONString(messageConstant));
    44.         return object;
    45.     }
    46.  
    47.     @Before("@annotation(kafkaProcess)")
    48.     public void doBefore(KafkaProcess kafkaProcess){
    49.         //推送消息--方法调用前
    50.         KafkaSingleDTO singleDTO = KafkaSingleUtil.getObject();
    51.         singleDTO.setAssetType(assetType);
    52.         String stageCode = StageCodeConstant.STAGE_CODE.get(singleDTO.getAssetType()).get(kafkaProcess.functionName());
    53.         KafkaSendMessageConstant messageConstant = new KafkaSendMessageConstant();
    54.         messageConstant.setStageCode(stageCode);
    55.         //...消息实体类 可自补充
    56.         log.info("资产类型{}方法前推送消息：{}", singleDTO.getAssetType(), JSON.toJSONString(messageConstant));
    57.     }
    58.  
    59.     private Object getParamValue(ProceedingJoinPoint joinPoint, String paramName) {
    60.         Object[] params = joinPoint.getArgs();
    61.         String[] parameterNames = ((MethodSignature) joinPoint.getSignature()).getParameterNames();
    62.  
    63.         if (parameterNames == null || parameterNames.length == 0) {
    64.             return null;
    65.         }
    66.         for (String param : parameterNames) {
    67.             if (param.equals(paramName)) {
    68.                 int index = ArrayUtils.indexOf(parameterNames, param);
    69.                 return params[index];
    70.             }
    71.         }
    72.         return null;
    73.     }
    74. }

    最终结果：

    到此结束！

  • 相关阅读:
    嵌入式分享合集83
    软件测试/测试开发丨明确的编码规范，避免冗余和混乱
    Tomcat最大并发数及在线用户数
    ASP.NET Core Web API 幂等性
    第九章 字符串处理函数
    Shiro框架 02（之认证）
    C++常见面试题汇总
    java计算机毕业设计西安市城市绿地管理系统源码+系统+数据库+lw文档
    supervisor进程管理 ——k8s从入门到高并发系列教程（十一）
    java面向对象-----再谈方法
  • 原文地址：https://blog.csdn.net/qq_37342720/article/details/133936538