TAF 必修课（四）：过载保护

public class ConnectionSessionListener implements SessionListener {

    private final AtomicInteger connStat = new AtomicInteger(0);

    private final int MaxConnCount;

    public ConnectionSessionListener(int connCount) {
        MaxConnCount = connCount;
        System.out.println("MaxConnCount=" + MaxConnCount);
    }

    @Override
    public synchronized void onSessionCreated(SessionEvent se) {
        System.out.println("onSessionCreated: " + connStat.get());
        
        if (connStat.get() >= MaxConnCount) {
            try {
                System.out.println("reached the max connection threshold, close the session.");
                se.getSession().close();
            }
            catch (IOException e) {
                e.printStackTrace();
            }

            return;
        }

        connStat.incrementAndGet();
    }

    @Override
    public synchronized void onSessionDestoryed(SessionEvent se) {
        System.out.println("onSessionDestoryed: " + connStat.get());
        if (se.getSession() != null && se.getSession().getStatus() == SessionStatus.SERVER_CONNECTED && connStat.get() > 0) {
            connStat.decrementAndGet();
        }        
    }

}

public synchronized void start() 
{
	if (started) return ;
	
	new Thread((Runnable)()->{
		System.out.println("[NAMI_CORE] The session manager service started...");
		
		long lastUpdateOperationTime = -1;
		
		while(true) 
		{
			try 					
			{
				for(Session session : sessionList) 
				{
					lastUpdateOperationTime = session.getLastOperationTime();
					//会话超时
					if ((System.currentTimeMillis() - lastUpdateOperationTime) > timeout) {
						String s = "The session has timed out. [from ip: " + session.getRemoteIp() + " port: "
										+ session.getRemotePort() + "]";
						if (Utils.getLogger() != null) {
							Utils.getLogger().warn(s);
						} else {
							System.out.println("[NAMI_CORE] " + s);
						}
						session.asyncClose();
						unregisterSession(session); 
					}
				}
				
				Thread.sleep(interval);
			}
			catch (Exception ex)
			{
			}
		}	
	}, "SessionManageImpl Thread").start();
	started = true;
}

// 更新会话的上次操作时间实现，即发生IO，省略无关代码
Iterator<SelectionKey> iter = selector.selectedKeys().iterator();				
while (iter.hasNext()) 
{
	SelectionKey key = iter.next();
	iter.remove();
	if( !key.isValid() ) continue;
 
	//1. Update the last operation time
	if (key.attachment() != null && key.attachment() instanceof Session) {
		((Session)key.attachment()).updateLastOperationTime();
	}
}

int connCount = 0;
for (Entry<String, ServantAdapterConfig> adapterConfigEntry : ConfigurationManager.getInstance()
		.getserverConfig().getServantAdapterConfMap().entrySet()) {
	if (OmConstants.AdminServant.equals(adapterConfigEntry.getKey())) {
		continue;
	}
	connCount += adapterConfigEntry.getValue().getMaxConns();
}
ConnectionSessionListener sessionListener = new ConnectionSessionListener(connCount);
sessionManager.addSessionListener(sessionListener);

private static synchronized Executor createExecutor(JceServiceRequest request) {
	String key = null, contextName = null, serviceName = null;
	Executor executor = null;

	if (request == null) return createDefaultExecutor(null);

	serviceName = request.getServantName();
	key = contextName + '_' + serviceName;

	executor = threadExecutors.get(key);
	if (executor != null) return executor;

	int minPoolSize = -1, maxPoolSize = -1, queueSize = -1;
	AppContainer container = ContainerManager.getContainer(AppContainer.class);
	AppContextImpl context = container.getDefaultAppContext();

	// 客户端请求错误的app或service时，返回默认的executor.
	if (context == null) {
		return getDefaultExecutor();
	}
	ServiceHomeSkeleton service = context.getCapHomeSkeleton(serviceName);
	if (service == null) {
		return getDefaultExecutor();
	}

	//设定队列长度，线程数（线程池大小）
	minPoolSize = service.getMinThreadPoolSize();
	maxPoolSize = service.getMaxThreadPoolSize();
	queueSize = service.getQueueSize();

	if (minPoolSize < 0 || maxPoolSize < 0 || queueSize < 0) {
		return createDefaultExecutor(key);
	}

	TaskQueue taskqueue = new TaskQueue(queueSize);
	TaskThreadPoolExecutor pool = new TaskThreadPoolExecutor(minPoolSize, maxPoolSize, 120, TimeUnit.SECONDS, taskqueue, new TaskThreadFactory("taserverThreadPool-exec-" + contextName + '.' + serviceName + '-'));
	taskqueue.setParent(pool);
	threadExecutors.put(key, pool);

	return pool;
}

//接收请求处理函数
public void readRequest() throws IOException
{
	Request request = null; IoBuffer tempBuffer = null;
	
	try {
		tempBuffer = readBuffer.duplicate().flip();
		
		while(true) 
		{
			tempBuffer.mark(); 
			
			if (tempBuffer.remaining() > 0)
			{
				request = selectorManager.getProtocolFactory().getDecoder().decodeRequest(tempBuffer, this);
			}
			else
			{
				request = null; 
			}
			
			if (request != null) {
				try {
					//设定请求产生时间
					request.resetBornTime();
					selectorManager.getThreadPool().execute(new WorkThread(request, selectorManager));
				} catch (Exception ex) {
					ex.printStackTrace();
				}
			} else {
				tempBuffer.reset();
				readBuffer = resetIoBuffer(tempBuffer);
				break;
			}
		}
	} catch (ProtocolException ex) {
		close();
		ex.printStackTrace();
	}
}

//业务线程处理函数，省略无关代码
if (req != null) //处理server端request请求
{   
	req.setProcessTime(System.currentTimeMillis());
	req.init();
	Response res = selectorManager.getProcessor().process(req, req.getIoSession());
	if (!res.isAsyncMode()) req.getIoSession().write(res);
}

long startTime = req.getProcessTime();

// 2. Directly to give up if exceeds the value of maxWaitingTimeInQueue
int maxWaitingTimeInQueue = ConfigurationManager.getInstance().getserverConfig().getServantAdapterConfMap().get(request.getServantName()).getQueueTimeout();
waitingTime = (int) (startTime - req.getBornTime());
if (waitingTime > maxWaitingTimeInQueue) {
	throw new InternalException("Wait too long, server busy.");
}

public Object invoke(Method method, Object... args) throws Exception {
	Object value = null;
	Context<JceServiceRequest, JceServiceResponse> context = null;
	try {
		context = ContextManager.getContext();
		preInvokeCapHomeSkeleton(context);

		value = method.invoke(this.service, fixParamValueType(method, args));
	} finally {
		if (!ContextManager.getContext().response().isAsyncMode()) {
			postInvokeCapHomeSkeleton(context);
		}
	}
	return value;
}
//发起调用前计数加一
public void preInvokeCapHomeSkeleton(Context<JceServiceRequest, JceServiceResponse> context) {
	if (this.maxLoadLimit == -1) {
		return;
	}
	this.invokeNumbers.incrementAndGet();
	if (this.invokeNumbers.intValue() > this.maxLoadLimit) {
		throw new RuntimeException(this.name + " is overload. limit=" + this.maxLoadLimit);
	}
}
//调用结束计数减一
public void postInvokeCapHomeSkeleton(Context<JceServiceRequest, ?> context) {
	if (this.maxLoadLimit == -1) {
		return;
	}
	this.invokeNumbers.decrementAndGet();
}