package com.walker.tcp.support;
|
|
import com.walker.infrastructure.ApplicationRuntimeException;
|
import com.walker.infrastructure.core.ApplicationBeanInitialized;
|
import com.walker.queue.AbstractQueueManager;
|
import com.walker.queue.QueueException;
|
import com.walker.tcp.ActionCallException;
|
import com.walker.tcp.ActionCallable;
|
import com.walker.tcp.ActionCallablePostProcessor;
|
import com.walker.tcp.Connection;
|
import com.walker.tcp.ConnectionManager;
|
import com.walker.tcp.Context;
|
import com.walker.tcp.Filter;
|
import com.walker.tcp.Request;
|
import com.walker.tcp.Response;
|
|
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;
|
|
/**
|
* 使用内存队列实现的队列管理器。</p>
|
* 在数据量不大的情况下可以这样使用，例如：接入设备只有几十台，并发量只有上百的情况下。<br>
|
* 如果数据量较大，可以采用redis或者mq队列实现方式。{ RedisQueueManager}
|
* @author 时克英
|
* @date 2018-11-28
|
*
|
*/
|
public class MemoryQueueManager extends AbstractQueueManager implements ApplicationBeanInitialized {
|
|
private final BlockingQueue<Request<?>> dataQueue = new ArrayBlockingQueue<>(2048);
|
|
private final List<Filter> filterList = new ArrayList<>();
|
private int filterCount = 0;
|
|
private ExecutorService service = Executors.newFixedThreadPool(maxWorkerThread);
|
private Runnable consumer = new DefaultConsumer();
|
|
public void addFilter(Filter filter){
|
for(Filter f : filterList){
|
if(filter.getName().equalsIgnoreCase(f.getName())){
|
throw new IllegalArgumentException("已经存在该过滤器，不能重复添加");
|
}
|
}
|
this.filterList.add(filter);
|
this.filterCount++;
|
}
|
|
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;
|
}
|
}
|
|
/**
|
// 处理业务（这样执行是单线程的）
|
// processAsync(context);
|
* */
|
|
// 使用ExecutorService的线程池来执行，而不是在这一个单线程中排队处理
|
service.execute(new ActionInvokeTask(context));
|
|
} catch (InterruptedException e) {
|
logger.error(DefaultConsumer.class.getName() + "消费者任务中断", e);
|
} catch (Exception e) {
|
logger.error("执行任务出现错误", e);
|
}
|
}
|
|
}
|
}
|
|
private class ActionInvokeTask implements Runnable{
|
|
private Context context;
|
|
public ActionInvokeTask(Context context){
|
this.context = context;
|
}
|
|
@Override
|
public void run() {
|
Request<?> request = context.getCurrentData();
|
ActionCallable action = ActionCallablePostProcessor.getAction(request.getProtocolNum());
|
if(action == null){
|
throw new ApplicationRuntimeException("action未定义，protocol = " + request.getProtocolNum());
|
}
|
|
try{
|
if(request.isRequireResponse()){
|
Response<?> response = action.action(request);
|
|
// 2018-09-20 时克英修改，响应对象中自动加入用户唯一编号
|
response.setName(request.getName());
|
|
Connection conn = connectionManager.getConnectionByName(request.getName());
|
if(conn == null || !conn.isConnected()){
|
processFailed(request);
|
return;
|
}
|
|
conn.write(response);
|
processSuccess(request);
|
|
} else {
|
action.action(request);
|
logger.debug("该请求不需要响应：" + request);
|
}
|
|
} catch(ActionCallException ex){
|
// 这里也可以加入提醒机制，待续
|
processFailed(request);
|
logger.error("执行action错误：" + ex.getMessage(), ex);
|
}
|
}
|
}
|
|
// /**
|
// * 业务Action执行方法。
|
// * @param context
|
// */
|
// protected void processAsync(Context context) throws Exception{}
|
|
@Override
|
public void push(String key, Object data, Object option) throws QueueException {
|
if(data == null){
|
throw new QueueException("request is required!");
|
}
|
if(!Request.class.isAssignableFrom(data.getClass())){
|
throw new QueueException("请求对象必须是: " + Request.class.getName());
|
}
|
|
try {
|
dataQueue.put((Request<?>)data);
|
logger.debug("加入一个数据到队列：" + data);
|
} catch (InterruptedException e) {
|
e.printStackTrace();
|
}
|
}
|
|
/**
|
* 消息处理失败，回调方法，子类可覆盖重写
|
* @param request
|
*/
|
protected void processFailed(Request<?> request){
|
logger.error("tcp消息处理失败，可能连接不存在或者已关闭，记录日志：" + request.toString());
|
}
|
|
/**
|
* 消息处理成功，回调方法，子类可覆盖重写
|
* @param request
|
*/
|
protected void processSuccess(Request<?> request){
|
logger.debug("tcp消息处理成功：" + request.getProtocolNum());
|
}
|
|
@Override
|
public void startup() {
|
if(this.connectionManager == null){
|
throw new IllegalArgumentException("connectionManager未配置");
|
}
|
logger.info("初始化TcpAction消费者任务：" + consumer.getClass().getName());
|
service.execute(consumer);
|
}
|
|
private ConnectionManager connectionManager;
|
|
public void setConnectionManager(ConnectionManager connectionManager) {
|
this.connectionManager = connectionManager;
|
}
|
|
}
|
