AOP日志记录之BlockingQueue队列实践

悠悠 2022-02-25 05:52 230阅读 0赞

## 前言 ##

因近期接到一个需求，需要对平台用户的行为进行记录，最常见的做法就是使用AOP，然后日志直接入库。这种做法在并发量不高的系统上是可行的，但当系统流量负载比较高时，日志直接入库势必会对系统性能造成一定的影响。因此今天将分享使用阻塞队列来实现一个简单的MQ来定时批量处理高并发下的海量日志。

### 什么是BlockingQueue？ ###

> 阻塞队列（BlockingQueue）是区别于普通队列多了两个附加操作的线程安全的队列。这两个附加的操作是：在队列为空时，获取元素的线程会等待队列变为非空。当队列满时，存储元素的线程会等待队列可用。阻塞队列常用于生产者和消费者的场景，生产者是往队列里添加元素的线程，消费者是从队列里拿元素的线程。阻塞队列就是生产者存放元素的容器，而消费者也只从容器里拿元素。

下面将介绍具体的实现步骤

### 1.创建一个系统日志切面类 ###

拦截接口请求，封装日志信息，放到队列中去

/** * 系统日志切面处理类 */
    @Aspect
    @Component
    public class WebLogAspect { 
    
    	private static final Logger logger = LoggerFactory.getLogger(WebLogAspect.class);
    	
    	ThreadLocal<SystemLogEntity> systemLogThreadLocal = new ThreadLocal<>();
    	
    	/** * 1.定义一个切入点，可以是一个规则表达式，比如下例中某个package下的所有函数，也可以是一个注解 * */
    	@Pointcut("@within(org.springframework.web.bind.annotation.RestController) || @within(org.springframework.stereotype.Controller)")
    	public void webLogCut() { 
    		
    	}
    	
    	/** * 前置通知, 在方法执行之前执行 * * @param joinPoint * @throws Throwable */
    	@Before("webLogCut()")
    	public void doBefore(JoinPoint joinPoint) throws Throwable { 
    		// 接收到请求，记录请求内容
    		ServletRequestAttributes attributes = (ServletRequestAttributes) RequestContextHolder.getRequestAttributes();
    		HttpServletRequest request = attributes.getRequest();
    		// 记录下请求内容
    		logger.info("URL: {}", request.getRequestURL().toString());
    		StringBuffer url = request.getRequestURL();
    		String requestDomain = url.delete(url.length() - request.getRequestURI().length(), url.length()).append("/").toString();
    		logger.info("DOMAIN: {}", requestDomain);
    		logger.info("IP: {}", IPAddressUtil.getClientIpAddress(request));
    		logger.info("HTTP_METHOD: {}", request.getMethod());
    		logger.info("CLASS_METHOD: {}", joinPoint.getSignature().getDeclaringTypeName() + "." + joinPoint.getSignature().getName());
    		logger.info("ARGS_ARRAY: {}", joinPoint.getArgs());
    		
    		StringBuilder params = new StringBuilder();
    		Enumeration<String> enums = request.getParameterNames();
    		while (enums.hasMoreElements()) { 
    			String name = enums.nextElement();
    			String value = request.getParameter(name);
    			params.append(name).append("=").append(request.getParameter(name)).append("&");
    			logger.info("REQUEST_PARAMETER: {}={}", name, value);
    		}
    		if (StringUtils.isEmpty(params.toString()) && joinPoint.getArgs().length > 0) { 
    			try { 
    				params.append(JSON.toJSONString(joinPoint.getArgs()[0]));
    			} catch (Exception e) { 
    			}
    		}
    		
    		long beginTime = System.currentTimeMillis();
    		SystemLogEntity systemLog = new SystemLogEntity();
    		systemLogThreadLocal.set(systemLog);
    		systemLog.setLogType(SystemLogTypeEnum.LOG_TYPE_1.getCode());
    		MethodSignature signature = (MethodSignature) joinPoint.getSignature();
    		systemLog.setRequestDomain(requestDomain); // 请求域名
    		systemLog.setRequestIp(IPAddressUtil.getClientIpAddress(request)); // ip地址
    		systemLog.setRequestUrl(request.getRequestURI()); // 请求地址
    		systemLog.setRequestMethod(className + "." + methodName + "()"); // 请求方法
    		systemLog.setRequestParams(params.toString()); // 请求参数
    		systemLog.setCreateTime(DateUtils.getNow());
    		systemLog.setSpendTime(new BigDecimal(String.valueOf(beginTime)));
    	}
    	
    	/** * 返回通知, 在方法返回结果之后执行 * * @param ret * @throws Throwable */
    	@AfterReturning(returning = "ret", pointcut = "webLogCut()")
    	public void doAfterReturning(Object ret) throws Throwable { 
    		// 处理完请求，返回内容
    		logger.info("RESPONSE: {}", ret);
    		SystemLogEntity systemLog = systemLogThreadLocal.get();
    		if(null != systemLog){ 
    			this.saveSystemLog();
    		}
    	}
    	
    	/** * 异常通知，当目标方法执行过程中出现异常时才会进行执行的代码 * * @param joinPoint * @param ex */
    	@AfterThrowing(throwing = "ex", pointcut = "webLogCut()")
    	public void afterthrowinglogging(JoinPoint joinPoint, Exception ex) { 
    		SystemLogEntity systemLog = systemLogThreadLocal.get();
    		if (null != systemLog) { 
    			this.saveSystemLog();
    		}
    	}
    
    	/** * 保存系统日志 */
    	private void saveSystemLog(){ 
    		SystemLogEntity systemLog = systemLogThreadLocal.get();
    		long beginTime = Long.valueOf(systemLog.getSpendTime().toString());
    		//执行时长(秒)
    		double spendTime = (System.currentTimeMillis() - beginTime) / 1000.0d;
    		logger.info("SPEND TIME(秒): {}",spendTime);
    		systemLog.setSpendTime(new BigDecimal(String.valueOf(spendTime)));
    		try { 
    			// 将系统日志放入到队列中分批处理
    			SystemLogQueue.getInstance().push(systemLog);
    		} catch (Exception e) { 
    			logger.error("添加队列失败,已超过队列最大长度：{}", e);
    		} finally { 
    			systemLogThreadLocal.remove();
    		}
    	}
    }

### 2.系统日志处理队列 ###

可将队列长度、默认处理长度、默认处理间隔时间配置到properties文件中，根据实际情况进行相应调整

@Component
    public class SystemLogQueue { 
    	
    	private final Logger logger = LoggerFactory.getLogger(SystemLogQueue.class);
    	
    	// 队列大小
    	public static int QUEUE_MAX_SIZE;
    	// 默认队列处理长度
    	public static int DEFAULT_HANDLE_SIZE;
    	// 默认间隔处理队列时间
    	public static int DEFAULT_HANDLE_TIME;
    	// 阻塞队列
    	private static BlockingQueue<SystemLogEntity> blockingQueue;
    	private static final SystemLogQueue systemLogQueue = new SystemLogQueue();
    	
    	@Value("${system.log.queue.maxsize:5000}")
    	private void setQueueMaxSize(int maxsize) { 
    		QUEUE_MAX_SIZE = maxsize;
    		logger.info("queueMaxSize：{}", QUEUE_MAX_SIZE);
    		blockingQueue = new ArrayBlockingQueue<SystemLogEntity>(QUEUE_MAX_SIZE);
    	}
    
    	@Value("${system.log.queue.default.handle.size:2000}")
    	private void setDefaultHandleSize(int handleSize) { 
    		DEFAULT_HANDLE_SIZE = handleSize;
    		logger.info("defaultHandleSize：{}", DEFAULT_HANDLE_SIZE);
    	}
    
    	@Value("${system.log.queue.default.handle.time:300000}")
    	private void setDefaultHandleTime(int handleTime) { 
    		DEFAULT_HANDLE_TIME = handleTime;
    		logger.info("defaultHandleTime：{}", DEFAULT_HANDLE_TIME);
    	}
    	
    	private SystemLogQueue() { 
    		
    	}
    
    	public static SystemLogQueue getInstance() { 
    		return systemLogQueue;
    	}
    
    	/** * 消息入队 * * @param systemLog * @return */
    	public boolean push(SystemLogEntity systemLog) { 
    		return SystemLogQueue.blockingQueue.add(systemLog);
    	}
    
    	/** * 消息出队 * * @return */
    	public SystemLogEntity poll() { 
    		SystemLogEntity result = null;
    		try { 
    			result = SystemLogQueue.blockingQueue.take();
    		} catch (InterruptedException e) { 
    			e.printStackTrace();
    		}
    		return result;
    	}
    
    	/** * 获取队列大小 * * @return */
    	public int size() { 
    		return SystemLogQueue.blockingQueue.size();
    	}
    }

### 3.系统日志批量处理接口 ###

/** * 队列数据批量处理接口 */
    public interface QueueProcess<T> { 
    	
    	void processData(List<T> list);
    	
    }

### 4.系统日志队列消费者 ###

/** * 系统日志消费者 */
    @Service
    @Transactional
    public class SystemLogQueueCustomer<T> implements QueueProcess<SystemLogEntity> { 
    
    	private final Logger logger = LoggerFactory.getLogger(SystemLogQueueCustomer.class);
    	
    	@Override
    	public void processData(List<SystemLogEntity> list) { 
    		if (null != list && !list.isEmpty()) { 
    			// 批量将日志入库
    			
    		}
    	}
    }

### 5.系统日志队列监听器 ###

此监听器将会根据配置的默认处理长度和处理间隔时间分批处理

/** * 系统日志队列监听（当队列达到一定数量和时间后处理队列数据 ） */
    @Component
    public class SystemLogQueueListener implements Runnable { 
    
    	private final Logger logger = LoggerFactory.getLogger(SystemLogQueueListener.class);
    
    	// 用来存放从队列拿出的数据
    	private List<SystemLogEntity> queueDataList;
    	// 回调接口
    	private QueueProcess<SystemLogEntity> process;
    
    	public SystemLogQueueListener(QueueProcess<SystemLogEntity> process) { 
    		this.process = process;
    		queueDataList = new ArrayList<SystemLogEntity>(SystemLogQueue.DEFAULT_HANDLE_SIZE);
    	}
    
    	@Override
    	public void run() { 
    		long startTime = System.currentTimeMillis();
    		SystemLogEntity t = null;
    		while (true) { 
    			try { 
    				SystemLogQueue systemLogQueue = SystemLogQueue.getInstance();
    				// 从队列拿出队列头部的元素，如果没有就阻塞
    				t = systemLogQueue.poll();
    				if (null != t) { 
    					queueDataList.add(t);
    				}
    				if (queueDataList.size() >= SystemLogQueue.DEFAULT_HANDLE_SIZE) { 
    					logger.info("队列数据超过默认处理长度，开始消费,size：{},queueSize：{}", queueDataList.size(), systemLogQueue.size());
    					startTime = batchProcess(queueDataList);
    					continue;
    				}
    				long currentTime = System.currentTimeMillis();
    				if (currentTime - startTime > SystemLogQueue.DEFAULT_HANDLE_TIME) { 
    					logger.info("超过队列默认处理间隔时间，开始消费,defaultTime：{},handleTime：{},size：{},queueSize：{}",
    							SystemLogQueue.DEFAULT_HANDLE_TIME, currentTime - startTime, queueDataList.size(), systemLogQueue.size());
    					startTime = batchProcess(queueDataList);
    					continue;
    				}
    				logger.info("未满足队列批量处理条件继续等待,defaultTime：{},handleTime：{},size：{},queueSize：{}",
    						SystemLogQueue.DEFAULT_HANDLE_TIME, currentTime - startTime, queueDataList.size(), systemLogQueue.size());
    			} catch (Exception e) { 
    				e.printStackTrace();
    			}
    		}
    	}
    
    	/** * 批量处理队列数据 * * @param queueDataList * @return */
    	private long batchProcess(List<SystemLogEntity> queueDataList) { 
    		// 处理队列
    		try { 
    			logger.info("批量日志消费开始,size：{}", queueDataList.size());
    			process.processData(queueDataList);
    			logger.info("批量日志消费结束,size：{}", queueDataList.size());
    		} catch (Exception e) { 
    			logger.info("批量日志消费异常,errorMsg：{},exception：{}", e.getMessage(), e);
    		} finally { 
    			// 清理掉dataList中的元素
    			this.clearQueueDataList();
    		}
    		return System.currentTimeMillis();
    	}
    
    	/** * 清理生成的list */
    	public void clearQueueDataList() { 
    		queueDataList = null;
    		queueDataList = new ArrayList<SystemLogEntity>();
    	}
    }

### 6.Spring boot环境下使用示例 ###

可在系统服务启动成功之后启动系统日志消费者线程进行日志消费

@Component
    public class SystemLogConsumerRunner implements CommandLineRunner{ 
    	
    	private final Logger logger = LoggerFactory.getLogger(SystemLogQueueCustomer.class);
    	@Autowired
    	private SystemLogQueueCustomer<SystemLogEntity> systemLogQueueCustomer;
    	
    	@Override
    	public void run(String... args) throws Exception { 
    		SystemLogQueueListener listener = new SystemLogQueueListener(systemLogQueueCustomer);
    		new Thread(listener).start();
    	}
    }

### 参考 ###

[https://blueyan.iteye.com/blog/2024478][https_blueyan.iteye.com_blog_2024478]

[http://www.kailing.pub/article/index/arcid/153.html][http_www.kailing.pub_article_index_arcid_153.html]

[https://blog.csdn.net/vernonzheng/article/details/8247564][https_blog.csdn.net_vernonzheng_article_details_8247564]

[https_blueyan.iteye.com_blog_2024478]: https://blueyan.iteye.com/blog/2024478
[http_www.kailing.pub_article_index_arcid_153.html]: http://www.kailing.pub/article/index/arcid/153.html
[https_blog.csdn.net_vernonzheng_article_details_8247564]: https://blog.csdn.net/vernonzheng/article/details/8247564