public abstract class AbsBatchProcess<T> implements Runnable {
// 记录条数
protected int recordNum;
// 空闲时间(ms)
protected long idleTime;
// 轮询间隔
protected long interval;
// 是否停止
private boolean stop;
// 数据
private Queue<T> data;
// 上次时间(ms)
private long lastTime;
// 监听
private SecScanf scanf;
public AbsBatchProcess() {
recordNum = 10;
idleTime = 5000;
interval = 1000;
stop = false;
data = new ConcurrentLinkedQueue<T>();
lastTime = System.currentTimeMillis();
scanf = new SecScanf();
}
/**
* 开始
*/
public final void start(){
stop = false;
scanf.stop = false;
new Thread(this).start();
new Thread(scanf).start();
}
/**
* 终止
*/
public final void stop(){
stop = true;
scanf.stop = true;
}
/* (non-Javadoc)
* @see java.lang.Runnable#run()
*/
@Override
public final void run() {
while(!stop){
synchronized (data) {
if(data.size() <= 0){
try {
data.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
T entity = null;
while((entity = data.poll()) != null){
process(entity);
}
lastTime = System.currentTimeMillis();
}
}
/**
* 处理实现
* @param entity
*/
protected abstract void process(T entity);
/**
* 添加要处理的数据
* @param entity
*/
public final void batch(T... entity){
if(entity != null){
data.addAll(Arrays.asList(entity));
}
}
private class SecScanf implements Runnable {
boolean stop = false;
@Override
public void run() {
while(!stop){
long current = System.currentTimeMillis();
if(data.size() >= recordNum || current - lastTime > idleTime){
synchronized (data) {
data.notify();
}
}
try {
Thread.sleep(interval);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
}