项目 - Redis消息队列+工作线程取出用户操作日志并入库（二）

1. 保存操作提至

1. 配置类

@ConditionalOnProperty(prefix = “spring.redis”, value = { “host”, “password” }) 注解表示只有application.properties中配置了spring.redis.host或者spring.redis.password配置类才会生效。

@Configuration
public class RestTemplateConfig {
    

    @Bean(name = "logRedisTemplate")
    @ConditionalOnProperty(prefix = "spring.redis", value = {
     "host", "password" })
    public RedisTemplate<String, Object> redisTempalte(RedisConnectionFactory connectionFactory) {
    
        RedisTemplate<String, Object> template = new RedisTemplate<>();
        template.setKeySerializer(new StringRedisSerializer());
        template.setValueSerializer(new StringRedisSerializer());
        template.setHashKeySerializer(new StringRedisSerializer());
        template.setHashValueSerializer(new StringRedisSerializer());
        template.setConnectionFactory(connectionFactory);
        template.afterPropertiesSet();
        return template;
    }
}

2. 利用线程池创建线程取出redis中的操作日志

@Component
public class LogProcessUtil {
    

    private static RedisTemplate<String, Object> redisTemplate;
    private static LogHandlerDelegate oprLogHandler;

    // 利用线程池创建2个线程
    private static final ThreadPoolExecutor THREADPOOL 
        = new ThreadPoolExecutor(
            2, 2, 0L,
            TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(),
            new NamedThreadFactory(LogProcessUtil.class.getSimpleName() + "-", true),
            new ThreadPoolExecutor.DiscardOldestPolicy()
    );

    // 依赖注入
    @Autowired
    public LogProcessUtil(
            @Qualifier("logRedisTemplate") RedisTemplate<String, Object> redisTemplate, 
            LogHandlerDelegate oprLogHandler
    ) {
    
        LogProcessUtil.redisTemplate = redisTemplate;
        LogProcessUtil.oprLogHandler = oprLogHandler;
    }

    /** * 监听消息队列 */
    @PostConstruct
    public static void listenMessage() {
    
        startOprLogHandler();
    }

    /** * 启动线程监听opr_log_channel消息队列的消息 */
    public static void startOprLogHandler() {
    
        THREADPOOL.execute(() -> {
    
            List<String> messages;
            while (true) {
    
                try {
    
                    messages = rightLeftMessages(Constants.RedisChannel.OPR_LOG_CHANEL, 1000);
                    if (!CollectionUtils.isEmpty(messages)) {
    
                        // 处理操作日志保存到数据库中
                        oprLogHandler.handleMessage(messages);
                    }
                } catch (Exception e) {
    
                    e.printStackTrace();
                }
            }
        });
    }

    /** * 批量读取redis中数据 * */
    public static List<String> rightLeftMessages(String key, long length) {
    
        List<String> result = new ArrayList<>();
        String value;
        try {
    
            for (int i = 0; i < length; i++) {
    
                value = (String) redisTemplate.opsForList().leftPop(key, 5, TimeUnit.SECONDS);
                if (StringUtils.isEmpty(value)) {
    
                    break;
                }
                result.add(value);
            }
            return result;
        } catch (Exception e) {
    
            e.getMessage();
        }
        return result;
    }

}

3. 保存日志到数据库 - Service层

public interface LogHandlerDelegate {
    
    /** * 日志数据处理 * * @param messages */
    void handleMessage(List<String> messages);
}

/** * 操作日志写入数据库的委托类 * */
@Slf4j
@Component(value = "oprLogHandler")
public class OprLogHandlerDelegate implements LogHandlerDelegate {
    

    private static ObjectMapper objectMapper = new ObjectMapper();
    private static final String DEFAULT_TIME_FORMAT = "yyyy-MM-dd HH:mm:ss";

    static {
    
        objectMapper.disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);
        objectMapper.disable(DeserializationFeature.ADJUST_DATES_TO_CONTEXT_TIME_ZONE);
        JavaTimeModule javaTimeModule = new JavaTimeModule();
        javaTimeModule.addSerializer(LocalDateTime.class, new LocalDateTimeSerializer(DateTimeFormatter.ofPattern(DEFAULT_TIME_FORMAT)));
        javaTimeModule.addDeserializer(LocalDateTime.class, new LocalDateTimeDeserializer(DateTimeFormatter.ofPattern(DEFAULT_TIME_FORMAT)));
        objectMapper.registerModule(javaTimeModule);
    }

    @Autowired
    private IOprLogDao oprLogDao;

    /** * 委托函数，由redis监听容器调用 * * @param messages 操作日志（json格式） */
    @Override
    public void handleMessage(List<String> messages) {
    
        List<OprLog> oprLogList = new ArrayList<>();
        messages.stream().forEach(message -> {
    
            oprLogList.add(jsonToClassInstance(message, OprLog.class));
        });
        saveToDatabase(oprLogList);
    }

    private <T> T jsonToClassInstance(String jsonBody, Class<T> clazz) {
    
        try {
    
            return objectMapper.readValue(jsonBody, clazz);
        } catch (JsonProcessingException e) {
    
            log.error("operation log json type cast to {} error: {}", clazz, e.getMessage());
            return null;
        }
    }

    private void saveToDatabase(List<OprLog> oprLogs) {
    
        try {
    
            oprLogDao.batchInsertOprLog(oprLogs);
        } catch (Exception e) {
    
            log.error("Insert failed");
        }
    }
}

4. 保存日志到数据库 - Dao层

@Mapper
public interface IOprLogDao {
    

    /** * 批量插入操作日志 * * @param list */
    void batchInsertOprLog(List<OprLog> list);

}

<insert id="batchInsertOprLog" parameterType="entity.OprLog">
    INSERT INTO `t_opr_log`
    (
    username,
    userIp,
    target,
    action,
    actionTime,
    result
    )
    VALUES
    <foreach collection ="list" item="oprlog" separator =",">
        (
        #{oprlog.username},
        #{oprlog.userIp},
        #{oprlog.target},
        #{oprlog.action},
        #{oprlog.actionTime},
        #{oprlog.result}
        )
    </foreach >
</insert>

5. 日志实体类

CREATE TABLE  IF NOT EXISTS `t_opr_log`  (
  `id` int UNSIGNED NOT NULL AUTO_INCREMENT COMMENT '操作id',
  `username` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL COMMENT '操作用户',
  `userIp` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT '' COMMENT '操作用户所在的ip,如果是后台定时任务的操作或者其他没有ip的操作,为null',
  `target` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL COMMENT '被操作对象',
  `action` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL COMMENT '操作动作',
  `actionTime` timestamp(0) NOT NULL DEFAULT CURRENT_TIMESTAMP(0) COMMENT '操作时间',
  `result` tinyint(1) NOT NULL COMMENT '结果,成功或者失败,1表示成功,0表示失败',

  PRIMARY KEY (`id`) USING BTREE
)ENGINE=InnoDB DEFAULT CHARSET=utf8;

@Data
@JsonIgnoreProperties(ignoreUnknown = true)
public class OprLog implements Serializable {
    

    private static final long serialVersionUID = 1L;

    /** * 操作id */
    @JsonIgnore
    private Integer id;

    /** * 操作用户 */
    private String username;

    /** * 操作用户所在的ip,如果是后台定时任务的操作或者其他没有ip的操作,为null */
    private String userIp;

    /** * 操作目标 */
    private String target;

    /** * 操作行为 */
    private String action;

    /** * 操作时间 */
    @JsonDeserialize(using = LocalDateTimeDeserializer.class)
    @JsonSerialize(using = LocalDateTimeSerializer.class)
    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8")
    private LocalDateTime actionTime;

    /** * 结果,成功或者失败,1表示成功,0表示失败 */
    private Boolean result;
}

6. 启动类

@SpringBootApplication
// 异步调用
@EnableAsync
public class PlatformApp {
    
    public static void main(String[] args) {
    
        SpringApplication.run(PlatformApp.class, args);
    }
}

2. 查询操作日志

1. Controller层

@RestController
@ResponseResult
@RequestMapping("/oprLog")
public class OprLogController {
    

    @Autowired
    private IOperationLogService operationLogService;

    @GetMapping("/search")
    public PageData<OprLog> searchOperationLog(@Valid OprLogQo oprLogQo) {
    
        return operationLogService.searchOperationLog(oprLogQo);
    }
}

2. Service层

public interface IOperationLogService {
    

    /** * 操作日志搜索 * * @param oprLogQo 搜索字段参数 * @return */
    PageData<OprLog> searchOperationLog(OprLogQo oprLogQo);
}

@Service
public class OperationLogServiceImpl implements IOperationLogService {
    

    @Autowired
    private IOprLogDao oprLogDao;

    private static final String ORDER_BY = "actionTime desc";

    @Override
    public PageData<OprLog> searchOperationLog(OprLogQo oprLogQo) {
    
        // 由于endTime只精确到天，且sql查询条件为左闭右开，所以endTime需要加一天
        oprLogQo.plusOneDayOnEndTime();

        // 分页插件
        PageHelper.startPage(oprLogQo.getPageNum(), oprLogQo.getPageSize(), ORDER_BY);
        PageInfo<OprLog> pageInfo = new PageInfo<>(oprLogDao.searchOperationLog(oprLogQo));

        return assembleOprLogPageData(pageInfo);
    }

    private PageData<OprLog> assembleOprLogPageData(PageInfo<OprLog> pageInfo) {
    
        PageData<OprLog> pageData = new PageData<>();

        pageData.setData(pageInfo.getList());
        pageData.setPageNum(pageInfo.getPageNum());
        pageData.setPageSize(pageInfo.getPageSize());
        pageData.setTotalCount(pageInfo.getTotal());

        return pageData;
    }
}

3. Dao层

@Mapper
public interface IOprLogDao {
    

    /** * 搜索操作日志 */
    List<OprLog> searchOperationLog(OprLogQo oprLogQo);
}

<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="dao.IOprLogDao">
    <sql id="table_opr_log">
        t_opr_log
    </sql>

    <resultMap id="BaseResultMap" type="entity.OprLog">
        <id column="id" property="id"/>
        <result column="username" property="username"/>
        <result column="userIp" property="userIp"/>
        <result column="target" property="target"/>
        <result column="action" property="action"/>
        <result column="actionTime" property="actionTime"/>
        <result column="result" property="result"/>
    </resultMap>

    <sql id="Base_Column_List">
        id, username, userIp, target, action, actionTime, result
    </sql>

    <select id="searchOperationLog" resultMap="BaseResultMap" parameterType="qo.OprLogQo">
        select
        <include refid="Base_Column_List"/>
        from
        <include refid="table_opr_log"/>
        where 1=1
        <if test="startTime != null">
            and actionTime &gt;= #{startTime}
        </if>
        <if test="endTime != null">
            and actionTime &lt; #{endTime}
        </if>
        <if test="result != null">
            and result=#{result}
        </if>
        <if test="userIp != null and userIp != ''">
            and userIp=#{userIp}
        </if>
        <if test="username != null and username != ''">
            and username like CONCAT('%', #{username}, '%')
        </if>
        <if test="target != null and target != ''">
            and target like CONCAT('%', #{target}, '%')
        </if>
        <if test="action != null and action != ''">
            and action like CONCAT('%', #{action}, '%')
        </if>
    </select>
</mapper>