写点什么

SpringCloud 网关实现线程池异步批量保存请求日志

作者:Java你猿哥
  • 2023-04-13
    湖南
  • 本文字数:6420 字

    阅读完需:约 21 分钟

本文章实现的是线程池异步批量保存请求日志,实现的是数据库中保存日志数据

日志过滤器添加

首先是在网关服务中添加日志过滤器

@Log4j2public class LogFilter implements GlobalFilter, Ordered {    private static final String START_TIME = "startTime";    private static final List<HttpMessageReader<?>> messageReaders = HandlerStrategies.withDefaults().messageReaders();
@Resource VisitRecordService visitRecordService;
@Override public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
ServerHttpRequest request = exchange.getRequest(); // 请求方法 HttpMethod method = request.getMethod(); // 请求头 HttpHeaders headers = request.getHeaders(); // 设置startTime 用来计算响应的时间 exchange.getAttributes().put(START_TIME, System.currentTimeMillis()); // 构建日志记录 AccessRecord accessRecord = visitRecordService.build(exchange);
if (method != null) { //设置请求方法 accessRecord.setMethod(method.name()); if (method == HttpMethod.GET) { //获取get请求参数 MultiValueMap<String, String> formData = request.getQueryParams(); if (!formData.isEmpty()) { //保存请求参数 accessRecord.setFormData(JSON.toJSONString(formData)); } } else if (method == HttpMethod.POST) { MediaType contentType = headers.getContentType(); if (contentType != null) { Mono<Void> voidMono = null; if (contentType.equals(MediaType.APPLICATION_JSON)) { // JSON voidMono = readBody(exchange, chain, accessRecord); } if (voidMono != null) { //计算请求时间 cacueConsumTime(exchange);
return voidMono; } } } }
visitRecordService.put(exchange, accessRecord); // 请求后执行保存 return chain.filter(exchange).then(saveRecord(exchange)); }
private Mono<Void> saveRecord(ServerWebExchange exchange) { return Mono.fromRunnable(() -> { cacueConsumTime(exchange); });
}
/** * 计算访问时间 * * @param exchange */ private void cacueConsumTime(ServerWebExchange exchange) { //请求开始时设置的自定义属性标识 Long startTime = exchange.getAttribute(START_TIME); Long consumingTime = 0L; if (startTime != null) { consumingTime = System.currentTimeMillis() - startTime; log.info(exchange.getRequest().getURI().getRawPath() + ": 耗时 " + consumingTime + "ms"); } visitRecordService.add(exchange, consumingTime); }

private Mono<Void> readBody(ServerWebExchange exchange, GatewayFilterChain chain, AccessRecord accessRecord) {
return DataBufferUtils.join(exchange.getRequest().getBody()).flatMap(dataBuffer -> {
byte[] bytes = new byte[dataBuffer.readableByteCount()]; dataBuffer.read(bytes); DataBufferUtils.release(dataBuffer); Flux<DataBuffer> cachedFlux = Flux.defer(() -> { DataBuffer buffer = exchange.getResponse().bufferFactory().wrap(bytes); DataBufferUtils.retain(buffer); return Mono.just(buffer); });

// 重写请求体,因为请求体数据只能被消费一次 ServerHttpRequest mutatedRequest = new ServerHttpRequestDecorator(exchange.getRequest()) { @Override public Flux<DataBuffer> getBody() { return cachedFlux; } };
ServerWebExchange mutatedExchange = exchange.mutate().request(mutatedRequest).build();
return ServerRequest.create(mutatedExchange, messageReaders) .bodyToMono(String.class) .doOnNext(objectValue -> { accessRecord.setBody(objectValue); visitRecordService.put(exchange, accessRecord); }).then( chain.filter(mutatedExchange) ); }); }
@Override public int getOrder() { return 2; } }
复制代码

HttpMessageReader 是用于读取 HTTP 消息的类。 ServerRequest.create 方法创建一个新的 ServerRequest 对象,该对象表示一个 HTTP 请求。 AccessRecord 是自定义的数据模型用来保存访问日志,代码如下:

import com.baomidou.mybatisplus.annotation.TableName;import lombok.AllArgsConstructor;import lombok.Data;import lombok.NoArgsConstructor;
import java.io.Serializable;import java.net.URI;import java.time.LocalDateTime;
@Data@AllArgsConstructor@NoArgsConstructor@TableName(value = "access_recoder_log")public class AccessRecord implements Serializable { private String formData; private URI targetUri; private String method; private String scheme; private String path; private String body; private String ip; private Integer status; private Long userId; private Long consumingTime; private LocalDateTime createTime;}
复制代码

VisitRecordService 就是我这里定义的异步保存日志的实现类

VisitRecordService 异步保存日志

ServerWebExchange 是 Spring WebFlux 中的一个接口,用于表示 HTTP 请求和响应的交换。它提供了访问请求和响应的方法,以及访问请求属性和响应属性的方法。可以使用它来处理 HTTP 请求和响应,例如修改请求头或响应体,或者将请求转发到另一个处理程序。

在过滤器的 filter 方法中获取到对应的 ServerWebExchange,再从其中读取访问信息

@Slf4j@Servicepublic class VisitRecordService {    //自定义的一个标识    private final String attributeKey = "visitRecord";    /**     * 构建一个 VisitRecord 实体类,但仅适用于获取 request 信息     *     * @param exchange gateway访问     * @return 访问信息     */    public AccessRecord build(ServerWebExchange exchange) {        // 获取请求信息        ServerHttpRequest request = exchange.getRequest();        String ip = RequestUtils.getIpAddress(request);        // 请求路径        String path = request.getPath().pathWithinApplication().value();        // 请求schema: http/https        String scheme = request.getURI().getScheme();        // 请求方法        HttpMethod method = request.getMethod();        // 路由服务地址        URI targetUri = exchange.getAttribute(ServerWebExchangeUtils.GATEWAY_REQUEST_URL_ATTR);        // 请求头        HttpHeaders headers = request.getHeaders();        // 获取请求地址        InetSocketAddress remoteAddress = request.getRemoteAddress();
AccessRecord accessRecord = new AccessRecord(); accessRecord.setPath(path); accessRecord.setScheme(scheme); accessRecord.setTargetUri(targetUri); accessRecord.setIp(ip); accessRecord.setCreateTime(LocalDateTime.now()); return accessRecord; } /** * 将访问信息存入 ServerWebExchange 当中,将会与当前请求关联起来, * 以便于后续在任何地方均可获得 * * @param exchange gateway访问合同 * @param visitRecord 访问信息 */ public void put(ServerWebExchange exchange, AccessRecord visitRecord) { Map<String, Object> attributes = exchange.getAttributes(); attributes.put(attributeKey, visitRecord); } }
复制代码

然后在请求结束的时候保存一下日志

@Slf4j@Servicepublic class VisitRecordService {    /**     * 缓存,在插入数据库前先存入此。     * 为防止数据被重复插入,故使用Set,但不能确保100%不会被重复存储。     */    private HashSet<AccessRecord> visitCache = new HashSet<>();    /**     * 保存访问记录     *     * @param exchange      gateway访问     * @param consumingTime  访问耗时     */    public void add(ServerWebExchange exchange, Long consumingTime) {        // 获取数据        ServerHttpResponse response = exchange.getResponse();        ServerHttpRequest request = exchange.getRequest();        //获取保存的日志记录体        AccessRecord visitRecord = getOrBuild(exchange);        //设置访问时间 单位毫秒        visitRecord.setConsumingTime(consumingTime);
// 设置访问状态 if (response.getStatusCode() != null) { visitRecord.setStatus(response.getStatusCode().value()); } //设置访问的用户ID 我这里是保存在请求头中 String userId = request.getHeaders().getFirst("userId"); if(StringUtils.isNotEmpty(userId)) { visitRecord.setUserId(Long.parseLong(userId)); } // 打印访问情况 log.info(visitRecord.toString()); // 添加记录到缓存中 visitCache.add(visitRecord); // 执行任务,保存数据 doTask(); }}
复制代码

doTask 在这里是使用线程池异步执行日志保存:

 /**     * 信号量,用于标记当前是否有任务正在执行,{@code true}表示当前无任务进行。     */    private volatile boolean taskFinish = true;    private final ThreadPoolExecutor threadPool = new ThreadPoolExecutor(1, 3, 15, TimeUnit.SECONDS, new ArrayBlockingQueue<>(10), namedThreadFactory, new ThreadPoolExecutor.AbortPolicy());    /**     * 单次批量插入的数据量     */    private final int BATCH_SIZE = 500;        private void doTask() {        if (taskFinish) {            // 当前没有任务的情况下,加锁并执行任务            synchronized (this) {                if (taskFinish) {                    taskFinish = false;                    threadPool.execute(() -> {                        try {                            // 当数据量较小时,则等待一段时间再插入数据,从而做到将数据尽可能的批量插入数据库                            if (visitCache.size() <= BATCH_SIZE) {                                Thread.sleep(500);                            }                            //批量保存                            batchSave();                        } catch (InterruptedException e) {                            log.error("睡眠时发生了异常: {}", e.getMessage());                        } finally {                            // 任务执行完毕后修改标志位                            taskFinish = true;                        }                    });                }            }        }    }
复制代码

ThreadPoolExecutor 是 Java 中的一个线程池实现,用于管理和复用线程,以提高应用程序的性能和响应能力。

它可以控制线程的数量,避免线程过多导致的资源浪费和性能下降,同时也可以避免线程不足导致的任务等待和响应延迟。

通过 ThreadPoolExecutor,我们可以将任务提交给线程池,由线程池中的线程来执行任务,从而实现任务的异步执行和并发处理。

然后 batchSave() 方法就是具体的实现数据保存:

@Slf4j@Servicepublic class VisitRecordService {    @Resource    VisitLogService visitLogService;    /**     * 缩减因子,每次更新缓存Set时缩小的倍数,对应HashSet的扩容倍数     */    private final float REDUCE_FACTOR = 0.5f;
private void batchSave() { log.debug("访问记录准备插入数据库,当前数据量:{}", visitCache.size()); if (visitCache.size() == 0) { return; } // 构造新对象来存储数据,旧对象保存到数据库后不再使用 HashSet<AccessRecord> oldCache = visitCache; visitCache = new HashSet<>((int) (oldCache.size() * REDUCE_FACTOR)); boolean isSave = false; try { //批量保存 isSave = visitLogService.saveBatch(oldCache, BATCH_SIZE); } finally { if (!isSave) { // 如果插入失败,则重新添加所有数据 visitCache.addAll(oldCache); } } } }
复制代码

VisitLogService 就是 mybatis 的正常的数据增删改查范畴,我这里的定义如下:

import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;import com.biglead.gateway.pojo.AccessRecord;import com.biglead.gateway.mapper.VisitLogMapper;import org.springframework.stereotype.Service;
/** * 访问日志Service类 */@Servicepublic class VisitLogService extends ServiceImpl<VisitLogMapper, AccessRecord> {
}
复制代码

VisitLogMapper 定义如下:

import com.baomidou.mybatisplus.core.mapper.BaseMapper;import com.biglead.gateway.pojo.AccessRecord;import org.apache.ibatis.annotations.Mapper;
@Mapperpublic interface VisitLogMapper extends BaseMapper<AccessRecord> {}
复制代码

然后启动项目,访问接口数据,就可以自动记录到数据库中


对应的 sql:

CREATE TABLE `access_recoder_log` (  `id` bigint(20) NOT NULL AUTO_INCREMENT,  `form_data` varchar(500) DEFAULT NULL,  `body` varchar(500) DEFAULT NULL,  `path` varchar(255) DEFAULT NULL,  `ip` varchar(255) DEFAULT NULL,  `status` int(11) DEFAULT NULL,  `user_id` bigint(20) DEFAULT NULL,  `scheme` varchar(255) DEFAULT NULL,  `method` varchar(255) DEFAULT NULL,  `consuming_time` bigint(20) DEFAULT NULL,  `create_time` datetime DEFAULT NULL,  PRIMARY KEY (`id`)) ENGINE=InnoDB AUTO_INCREMENT=25 DEFAULT CHARSET=utf8 COMMENT='日志记录表';
复制代码

最后就是源码了:https://gitee.com/android.long/spring-cloud-biglead/tree/master/biglead-api-11-admin


用户头像

Java你猿哥

关注

一只在编程路上渐行渐远的程序猿 2023-03-09 加入

关注我,了解更多Java、架构、Spring等知识

评论

发布
暂无评论
SpringCloud 网关实现线程池异步批量保存请求日志_spring_Java你猿哥_InfoQ写作社区