package com.walker.tcp.websocket;
|
|
import java.util.ArrayList;
|
import java.util.List;
|
import java.util.concurrent.ArrayBlockingQueue;
|
import java.util.concurrent.BlockingQueue;
|
import java.util.concurrent.ExecutorService;
|
import java.util.concurrent.Executors;
|
|
import com.walker.tcp.Context;
|
import com.walker.tcp.Filter;
|
import com.walker.tcp.Request;
|
import com.walker.tcp.Response;
|
import com.walker.tcp.support.AbstractActionCall;
|
import com.walker.tcp.support.SimpleContext;
|
|
/**
|
* 该对象虽然能够通过内部消息队列阻塞处理，但因在action内部，所以只能处理一种类型请求。
|
* 所以存在较大局限性，需要设计新的对象来统一处理消息队列任务。
|
* @author 时克英
|
* @date 2018-09-15
|
*
|
*/
|
@Deprecated
|
public abstract class DefaultProcessAction extends AbstractActionCall{
|
|
private final List<Filter> filterList = new ArrayList<>();
|
private int filterCount = 0;
|
|
private final BlockingQueue<Request<?>> dataQueue = new ArrayBlockingQueue<>(1024);
|
|
private ExecutorService service = Executors.newFixedThreadPool(1);
|
private Runnable consumer = new DefaultConsumer();
|
|
public void init(){
|
logger.info("初始化TcpAction消费者任务：" + consumer.getClass().getName());
|
service.execute(consumer);
|
}
|
|
public void addFilter(Filter filter){
|
for(Filter f : filterList){
|
if(filter.getName().equalsIgnoreCase(f.getName())){
|
throw new IllegalArgumentException("已经存在该过滤器，不能重复添加");
|
}
|
}
|
this.filterList.add(filter);
|
this.filterCount++;
|
}
|
|
@Override
|
public Response<?> action(Request<?> request) {
|
if(request == null){
|
throw new IllegalArgumentException("request is required!");
|
}
|
|
if(request.isRequireResponse()){
|
return doProcessSync(request);
|
}
|
|
/* 把数据存入队列，排队处理 */
|
try {
|
dataQueue.put(request);
|
logger.debug("******** 加入一个数据到队列：" + request);
|
} catch (InterruptedException e) {
|
e.printStackTrace();
|
}
|
return null;
|
}
|
|
/**
|
* 对于请求需要返回响应的数据，只能在同步方法<code>action</code>中处理
|
* @param request
|
* @return
|
*/
|
protected abstract Response<?> doProcessSync(Request<?> request);
|
|
/**
|
* 对于不需要响应的请求，可以在异步线程中处理，通过消息队列来排队执行。
|
* @param context
|
* @return
|
*/
|
protected abstract Response<?> doProcessASync(Context context);
|
|
private class DefaultConsumer implements Runnable{
|
@Override
|
public void run() {
|
logger.info("处理websocket线程启动：" + this.getClass().getName());
|
while(true){
|
try {
|
Request<?> request = dataQueue.take();
|
Context context = new SimpleContext(request, dataQueue.peek());
|
logger.debug("消费者拿到一个数据：" + request);
|
|
if(filterCount > 0){
|
/* 存在过滤器，先过滤 */
|
boolean filterResult = true;
|
for(Filter f : filterList){
|
filterResult = f.doFilter(context);
|
if(!filterResult){
|
logger.debug("过滤器终止操作：" + f.getName());
|
break;
|
}
|
}
|
if(!filterResult){
|
return;
|
}
|
}
|
|
// 处理业务
|
doProcessASync(context);
|
|
} catch (InterruptedException e) {
|
logger.error(DefaultConsumer.class.getName() + "消费者任务中断", e);
|
} catch (Exception e) {
|
logger.error("执行任务出现错误", e);
|
}
|
}
|
|
}
|
}
|
|
}
|
