场景描述
在项目中有这样一个需求,用户下载app并打开进入首页之前,会让用户选择自己感兴趣的主题分类,后台根据用户的选择提取出用户的行为数据并作出统计,以图形或者表格的形式展现出来,后期就可以根据这些数据做去做一些类似个性化或精准化的推送了,例如在进入豆瓣app首页前提示用户先选择感兴趣的主题内容.
image.png
具体实现
由于项目中主要使用了Spring框架,所以首先想到的最简单的方式就是使用aop切面拦截用户选择主题分类的接口,然后在数据库中建一张单独的表t_user_interest来存储用户的行为数据,由于是统计型数据所以实时性要求不高,允许有一定的延迟,并且为避免频繁的操作数据库,在切面拦截数据后起独立的线程推送到redis队列,后台再起调度任务每隔1小时异步持久化到数据库中.下面来看看实现步骤:
1.定义aop拦截器拦截方法调用,把用户行为数据推送到redis队列中,其中TaskPo为封装好的与用户相关的行为数据对象.
/**
* aop拦截器拦截方法调用
*/
@Component
@Aspect
public class UserInterestAspect {
private final static Logger logger = LoggerFactory.getLogger(UserInterestAspect.class);
private final LinkedBlockingQueue<TaskPo> queue = new LinkedBlockingQueue<>();
private final static ExecutorService pool = Executors.newFixedThreadPool(5);
private final static String QUEUE_NAME = "USER_INTEREST";
/**
* 前置通知:在目标方法开始之前执行
*
* @param joinPoint
* @throws Exception
*/
@Before("execution(* xxx.service.impl.xxxServiceImpl.xxx(..))")
public void saveRequiresLog(JoinPoint joinPoint) throws Exception {
try {
TaskPo taskPo = getTaskPo(joinPoint);
if (taskPo != null) {
queue.add(taskPo);
TaskPo taskMsge;
while ((taskMsge = queue.poll()) != null) {
pool.execute(new PushRedisWorker(QUEUE_NAME, taskMsge));
}
}
} catch (Exception e) {
throw e;
}
}
/**
* 解析参数映射为Java对象
*
* @param joinPoint
* @return
*/
public TaskPo getTaskPo(JoinPoint joinPoint) {
Object[] paramsValue = joinPoint.getArgs();
String[] paramsName = ((CodeSignature) joinPoint.getStaticPart()
.getSignature()).getParameterNames();
JSONObject entityJSON = new JSONObject();
for (int i = 0; i < paramsName.length; i++) {
entityJSON.put(paramsName[i], paramsValue[i]);
}
if (entityJSON != null && !entityJSON.isEmpty()) {
TaskPo taskPo = JSONObject.toJavaObject(entityJSON, TaskPo.class);
taskPo.setOperatorDate(new Date());
return taskPo;
}
return null;
}
}
2.接着起一个任务线程接口worker继承Runnable.
public interface Worker extends Runnable {
}
3.然后创建一个PushRedisWorker类实现Worker接口,并将封装好的数据写到redis队列中.
public class PushRedisWorker implements Worker {
private String QUEUE_NAME;
private TaskPo taskPo;
public PushRedisWorker(String QUEUE_NAME, TaskPo taskPo) {
this.QUEUE_NAME = QUEUE_NAME;
this.taskPo = taskPo;
}
@Override
public void run() {
// TODO Auto-generated method stub
try (Jedis jedis = JedisUtils.getJedis()) {
jedis.lpush(QUEUE_NAME, JSON.toJSONString(taskPo));
}
}
}
4.然后再定义一个调度任务器类每隔一段时间间隔轮询redis任务队列,不断的从队列中消费数据,再结合线程池异步发送数据到mysql中.
/**
* 任务调度定时到Redis队列中拉取对象持久化到数据库
*/
@Component
public class RedisTaskJob {
private final static Logger logger = LoggerFactory.getLogger(RedisTaskJob.class);
private final static String QUEUE_NAME = "USER_INTEREST";
private final static ExecutorService es = Executors.newFixedThreadPool(5);
private volatile static boolean isRun = true;
/**
* 每隔一小时执行一次
*/
@Scheduled(cron = "0 0 0/1 * * ? ")
public void getRedisTask() {
if (logger.isDebugEnabled()) {
logger.debug("调度开始");
}
//取出spring受管的业务service实例
IUserInterestService userInterestService = SpringContextHolder.getBean("userInterestService",IUserInterestService.class);
try (Jedis jedis = JedisUtils.getJedis()) {
if (jedis.exists(QUEUE_NAME)) {
start();
}
while (isRun) {
if (!jedis.exists(QUEUE_NAME)) {
stop();
break;
}
try {
String task = jedis.lpop(QUEUE_NAME);
if (StringHelpUtils.isNotBlank(task)) {
TaskPo taskPo = JSONObject.toJavaObject(JSON.parseObject(task), TaskPo.class);
//任务对象转化为实体model
UserInterest model = EntityUtils.convert(taskPo,UserInterest.class);
es.submit(() -> {
//执行数据入库操作
userInterestService.save(model);
});
}
} catch (Exception e) {
stop();
throw e;
}
}
}
}
public static void stop() {
isRun = false;
}
public static void start() {
isRun = true;
}
}
5.最后统计mysql中的数据并生成报表.
