package com.javaweb.cms.service;
|
|
import cn.hutool.core.collection.CollectionUtil;
|
import com.google.common.collect.Queues;
|
import com.javaweb.cms.domain.Pv;
|
import com.javaweb.common.utils.IpUtils;
|
import com.javaweb.common.utils.StringUtils;
|
import com.javaweb.common.utils.useragent.UserAgentUtils;
|
import eu.bitwalker.useragentutils.Browser;
|
import eu.bitwalker.useragentutils.DeviceType;
|
import org.apache.logging.log4j.LogManager;
|
import org.apache.logging.log4j.Logger;
|
import org.springframework.beans.factory.InitializingBean;
|
import org.springframework.beans.factory.annotation.Autowired;
|
import org.springframework.stereotype.Component;
|
|
import javax.annotation.PreDestroy;
|
import javax.servlet.http.HttpServletRequest;
|
import java.util.ArrayList;
|
import java.util.List;
|
import java.util.concurrent.*;
|
|
@Component("pvQueueService")
|
public class PVQueueService implements InitializingBean {
|
|
private static final Logger logger = LogManager.getLogger(PVQueueService.class);
|
|
@Autowired
|
IPvService pvService;
|
|
//创建一个可重用固定线程数的线程池
|
private ExecutorService pool = Executors.newFixedThreadPool(1);
|
|
private static final BlockingQueue<Pv> blockingQueue = new ArrayBlockingQueue<Pv>(1000000);
|
|
//线程活动
|
private volatile boolean threadActivity = true;
|
|
@Override
|
public void afterPropertiesSet() throws Exception {
|
pool.execute(new Runnable(){
|
@Override
|
public void run() {
|
while (threadActivity) { //如果系统关闭，则不再运行
|
try {
|
List<Pv> data = new ArrayList<Pv>();
|
//每次到1000条数据才进行入库，或者等待1分钟，没达到100条也继续入库
|
Queues.drain(blockingQueue, data, 50, 1, TimeUnit.MINUTES);//第三个参数：数量; 第四个参数：时间; 第五个参数：时间单位
|
if(CollectionUtil.isNotEmpty(data)){
|
pvService.insertPvBatch(data);
|
}
|
|
} catch (InterruptedException e) {
|
// e.printStackTrace();
|
if (logger.isErrorEnabled()) {
|
logger.error("访问量消费队列错误",e);
|
}
|
}
|
}
|
}});
|
}
|
|
public void pushPvQueue(HttpServletRequest request,Pv pv){
|
|
pv.setIp(IpUtils.getIpAddr(request));
|
if(StringUtils.isEmpty(pv.getReferer())){
|
pv.setReferer(request.getHeader("referer"));
|
}
|
Browser browserObj = UserAgentUtils.getBrowser(request);
|
String browser=browserObj.getName();//浏览器类型
|
pv.setBrowser(browser);
|
String deviceType="Unknown";//设备类型
|
DeviceType deviceType1=UserAgentUtils.getDeviceType(request);//是否pc
|
if(deviceType1!=null){
|
deviceType=deviceType1.getName();
|
}
|
pv.setDeviceType(deviceType);
|
|
//add(anObject):添加元素到队列里，添加成功返回true，容量满了添加失败会抛出IllegalStateException异常
|
//offer(anObject):表示如果可能的话,将anObject加到BlockingQueue里,即如果BlockingQueue可以容纳,则返回true,否则返回false.（本方法不阻塞当前执行方法的线程）
|
//offer(E o, long timeout, TimeUnit unit),可以设定等待的时间，如果在指定的时间内，还不能往队列中加入BlockingQueue，则返回失败。
|
//put(anObject):把anObject加到BlockingQueue里,如果BlockQueue没有空间,则调用此方法的线程被阻断直到BlockingQueue里面有空间再继续.
|
blockingQueue.offer(pv);//添加一个元素并返回true 如果队列已满，则返回false
|
}
|
@PreDestroy
|
public void destroy() {
|
threadActivity = false;
|
pool.shutdownNow();
|
}
|
}
|
