广告调度中心

背景介绍

      一支广告的投放通常有很多的限定条件：地域、时段、频次、广告位、轮播顺序、用户标签【性别、年龄、爱好】。广告主的期望有良好的投放效果，广告平台期望有良好的投放量以及效果。假设媒体资源足够，那我们应该需要计算出每一分钟应该完成的投放量，调度就是为了完成这个目标。

 

调度需要的数据

• 容量曲线
• 分钟级别的排期
• 总的投放量
• 已经完成的数据量

 

计算公式

• 本分钟应该完成的量=(总的投放量-已完成量)*容量因子
• 容量因子=当前分钟容量/剩余排期容量 [计算过程数据完全依赖容量曲线]

 

 

已经完成的量

      获取这个数据的方式有很多种方式，我们这里通过广告投放机器提供的接口获得，调度中心只按照自己的策略获取即可，由于投放机器有很多我们需要并发获取数据来加快整个调度的过程。通过CountDownLatch来控制是否完成了所有机器都的收账。

 

收账代码

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;

import net.sf.json.JSONArray;
import net.sf.json.JSONException;
import net.sf.json.JSONObject;

import org.apache.http.HttpEntity;
import org.apache.http.HttpResponse;
import org.apache.http.client.ClientProtocolException;
import org.apache.http.util.EntityUtils;
import org.slf4j.Logger;

import com.javagc.entity.ErrorStackMessage;
import com.javagc.entity.MergePolicy;
import com.javagc.context.CollectEffectDataContext;
import com.javagc.context.ServerGroupContext.GROUPS;
import com.javagc.entity.CollectEffectData;
import com.javagc.entity.CollectRunStatus;
import com.javagc.entity.CollectTaskDTO;
import com.javagc.entity.CurrCastData.DELIVER_TYPE;
import com.javagc.service.IEffectCollectService;
import com.javagc.util.DispatcherConstants;
import com.javagc.util.DispatcherUtil;

/**
 * 任务 stevenDing<br>
 */
public class CollectTask implements Runnable {

	private static final Logger logger = DispatcherUtil.getLog();
	
	private CollectEffectData 			  collectEffectData;
	private Map<String, CollectRunStatus> runStatusMap;
	private CountDownLatch 		 latch;
	private IEffectCollectService collect;
	private CollectTaskDTO 		 taskDTO;
	private GROUPS  			 group;

	// shared data
	private int 		generateJsonCost;
	private String 		result;
	private boolean 	runStatus;
	private long 		endTime;
	private List<Long> 	reTryRunTimeList = new ArrayList<Long>();
	
	/**
	 * 
	 * @param taskDTO
	 * @param effectData
	 * @param runLogMap
	 * @param latch
	 * @param collect
	 */
	public EffectCollectTask(	CollectTaskDTO 		taskDTO, 
								CollectEffectData 	collectEffectData,
								Map<String, CollectRunStatus> runStatusMap, 
								CountDownLatch 			latch,
								IEffectCollectService 	collect,
								GROUPS  group) {
		
		this.taskDTO 			= taskDTO;
		this.collectEffectData 	= collectEffectData;
		this.runStatusMap 		= runStatusMap;
		this.latch 				= latch;
		this.collect 			= collect;
		this.group				= group;
	}

	public void run() {

		try {
			// retry it
			while (true) {
				// invoke effect data
				long retryBegin = System.currentTimeMillis();
				try {
					invokeEffectData();
					runStatus = true;
				} catch (Throwable e) {
					runStatus = false;
					logger.error(e.getMessage(), e);
					ErrorStackMessage msg = new ErrorStackMessage(e, "", new MergePolicy(false,true));
					DispatcherConstants.collectErrorStackCollector.produce(msg);
				} finally {
					reTryRunTimeList.add(System.currentTimeMillis() - retryBegin);
				}
				
				//任务剩余时间
				long remainingTime = (taskDTO.getBeginTime() + taskDTO.getTimeout()) - System.currentTimeMillis();

				//成功或剩余时间不足则退出收账
				if (isRunStatus()|| 
					remainingTime < DispatcherConstants.COLLECTOR_BREAKTIME) {
					break;
				}else{
					try{
						Thread.sleep(DispatcherConstants.COLLECTOR_RETRYSLEEPMS);
					}catch(Exception e){}
				}
			}
		} finally {
			try {
				if (taskDTO.getClient() != null) {
					taskDTO.getClient().getConnectionManager().shutdown();
				}
			} catch (Throwable e) {
				logger.error(e.getMessage(), e);
				ErrorStackMessage msg = new ErrorStackMessage(e, "", new MergePolicy(false,true));
				DispatcherConstants.collectErrorStackCollector.produce(msg);
			}
		}

		try {
			// invoke succ
			if (isRunStatus()) {
				parseEffectData();
				CollectEffectDataContext.I.updateCollectEffectData(group,collectEffectData);
				runStatus = true;
			}
		} catch (Throwable e) {
			logger.error(e.getMessage(), e);
			ErrorStackMessage msg = new ErrorStackMessage(e, "", new MergePolicy(false,true));
			DispatcherConstants.collectErrorStackCollector.produce(msg);
			runStatus = false;
		} finally {
			try {
				endTime = System.currentTimeMillis();
				CollectRunStatus runStatus = wrapperRunStatus();
				// check main thread status
				if (!collect.isCollectCompleted()) {
					runStatusMap.put(taskDTO.getIp(), runStatus);
				}
			} catch (Throwable e) {
				logger.error(e.getMessage(), e);
				ErrorStackMessage msg = new ErrorStackMessage(e, "", new MergePolicy(false,true));
				DispatcherConstants.collectErrorStackCollector.produce(msg);
			}
			latch.countDown();
		}
	}

	protected void invokeEffectData() throws ClientProtocolException, IOException {
		HttpResponse response = taskDTO.getClient().execute(taskDTO.getGet());
		HttpEntity entity = response.getEntity();
		if (entity != null) {
			result = EntityUtils.toString(entity, "UTF-8");
		}
	}

	/**
	 * json --> {"cost" : 34,"s" : [{"id" : 111, "t" : 58},{"id" : 222,"t" : 66}],
	 * "c" : [{"id" : 10, "t" : 23},{"id" : 15,"t" : 39}],"vs" : [{"id" : 20, "t" : 13},{"id" : 45,"t" : 22}],
	 *   "ps" : [{"id" : 30, "t" : 42},{"id" : 37,"t" : 18}]} 
	 */
	protected void parseEffectData() throws Exception {
		
		JSONObject object = JSONObject.fromObject(getResult());
		// cost time
		generateJsonCost = object.getInt("cost");		
		parseEffectData(DELIVER_TYPE.CPM, object, "s");
		parseEffectData(DELIVER_TYPE.CPC, object, "c");
		parseEffectData(DELIVER_TYPE.CPV, object, "vs");
		parseEffectData(DELIVER_TYPE.CPP, object, "ps");
	}
	
	/**
	 * 
	 * @param type
	 * @param jsonArray
	 */
	protected  void parseEffectData(DELIVER_TYPE type,JSONObject object,String nodeName){
		
		Map<Long, Long> effectMap = new HashMap<Long, Long>();
		try{
			//曝光数据
			JSONArray jsonArray = object.getJSONArray(nodeName);
			//效果数据
			for (int i = 0; i < jsonArray.size(); i++) {
				JSONObject data = (JSONObject) jsonArray.get(i);
				try{
					if(isValidValues(data.getString("t"))){
						long showTimes = data.getLong("t");
						if (showTimes >= 0) {
							effectMap.put(data.getLong("id"), showTimes);	
						}
					}
				}catch(JSONException e){
					StringBuffer buffer = new StringBuffer();
					buffer.append(type+" Effect Data, JSONException, jsonObject=");
					buffer.append(data.toString()).append(" , ip=").append(taskDTO.getIp());
					logger.error(buffer.toString(),e);
					ErrorStackMessage msg = new ErrorStackMessage(e, buffer.toString(), new MergePolicy(true,true));
					DispatcherConstants.collectErrorStackCollector.produce(msg);
				}
			}
	
			if (!collect.isCollectCompleted()) {
				// put result to map
				collectEffectData.putEffectData(taskDTO.getIp(), effectMap,type);
			}
			
		}catch(Exception e){
			if(nodeName.equals("s") || nodeName.equals("c")){
				StringBuffer buffer = new StringBuffer();
				buffer.append(type+" Effect Data, JSONException, ps=null");
				logger.error(buffer.toString(),e);
				ErrorStackMessage msg = new ErrorStackMessage(e, buffer.toString(), new MergePolicy(true,true));
				DispatcherConstants.collectErrorStackCollector.produce(msg);
			}
		}
	}
	
	/**
	 * 包装运行状态
	 * @return
	 */
	protected CollectRunStatus wrapperRunStatus() {
		CollectRunStatus runStatus = new CollectRunStatus();
		runStatus.setIp(taskDTO.getIp());
		runStatus.setGenerateJsonCost(getGenerateJsonCost());
		runStatus.setThreadRunTime(getEndTime() - taskDTO.getBeginTime());
		runStatus.setRunStatus(isRunStatus());
		runStatus.setReTryRunTimeList(getReTryRunTimeList());
		return runStatus;
	}
	
	/**
	 * 检查数据是否有效
	 * @param content
	 * @return
	 */
	private boolean isValidValues(String content){
		
		if(content==null || content.length()==0 || content.getBytes().length>12){
			return false;
		}
		return true;
	}
	
	public List<Long> getReTryRunTimeList() {
		return reTryRunTimeList;
	}

	public long getEndTime() {
		return endTime;
	}

	public boolean isRunStatus() {
		return runStatus;
	}

	public String getResult() {
		return result;
	}

	public int getGenerateJsonCost() {
		return generateJsonCost;
	}
}

 

主进程控制

import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.http.Header;
import org.apache.http.HeaderElement;
import org.apache.http.HttpEntity;
import org.apache.http.HttpException;
import org.apache.http.HttpRequest;
import org.apache.http.HttpRequestInterceptor;
import org.apache.http.HttpResponse;
import org.apache.http.HttpResponseInterceptor;
import org.apache.http.client.HttpClient;
import org.apache.http.client.entity.GzipDecompressingEntity;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.client.DefaultHttpClient;
import org.apache.http.params.CoreConnectionPNames;
import org.apache.http.protocol.HttpContext;
import org.slf4j.Logger;

import com.javagc.context.ServerGroupContext;
import com.javagc.context.ServerGroupContext.GROUPS;
import com.javagc.entity.CollectRunStatus;
import com.javagc.entity.CollectEffectData;
import com.javagc.entity.CollectTaskDTO;
import com.javagc.task.EffectCollectTask;
import com.javagc.util.DispatcherConstants;
import com.javagc.util.DispatcherUtil;

/**
 * 收账服务<br>
 * 1、一组服务器收账使用最大时长,@see DispatcherConstants.COLLECTOR_TOTALTIMEOUT
 * @author stevending<br>
 *
 */
public class CollectService implements ICollectService {

	private static final Logger logger = DispatcherUtil.getLog();
	
	/**
	 * 分组信息
	 */
	private GROUPS groupNum;
	/**
	 * 收账结果日志
	 */
	private Map<String,CollectRunStatus> runStatusMap = null;
	/**
	 * 结果数据
	 */
	private CollectEffectData 		effectData 	= null;
	/**
	 * 检测是否完成收账
	 */
	private CountDownLatch 	latch 		= null;
	/**
	 * 收账完成状态 true=完成、false=未完成
	 */
	private AtomicBoolean	collectCompleted = new AtomicBoolean(false);
	
	/**
	 * 构造方法
	 * @param groupNum
	 */
	public EffectCollectService(GROUPS groupNum){
		this.groupNum = groupNum;
	}
	
	
	@Override
	public CollectEffectData doCollect() throws InterruptedException{
		
		long beginTime = System.currentTimeMillis();
		try{
			//exec task
			exeTask();
		} finally{
			collectCompleted.set(true);
			//write run status
			writeRunStatus();
		}
		logger.info(groupNum+" Total runTime:"+(System.currentTimeMillis()-beginTime) + " ms");
		return effectData;
	}
	
	/**
	 * 执行任务
	 * @throws InterruptedException
	 */
	private void exeTask() throws InterruptedException{
		
		long beginTime		= System.currentTimeMillis();
		List<String> ips 	= ServerGroupContext.I.getGroupServerIps(groupNum);
		latch		=	new CountDownLatch(ips.size());
		runStatusMap 	= 	new ConcurrentHashMap<String,CollectRunStatus>();
		effectData 	=	new CollectEffectData();
		
		for(int i=0;i<ips.size();i++){
			CollectTaskDTO staskDTO = new CollectTaskDTO( getHttpClient(),
														getHttpGet(ips.get(i),DispatcherConstants.COLLECTOR_PORT),
														ips.get(i),
														DispatcherConstants.COLLECTOR_TOTALTIMEOUT,
														beginTime);
			CollectTask stask = new CollectTask(staskDTO,effectData,runStatusMap,latch,this,groupNum);
			Thread st = new Thread(stask);
			st.setDaemon(true);
			st.start();
		}
		//等待收账完成
		latch.await(DispatcherConstants.COLLECTOR_TOTALTIMEOUT, TimeUnit.MILLISECONDS);
		
		//TODO 清理未完成的线程
	}
	
	/**
	 * 记录运行状态
	 */
	protected void writeRunStatus(){
		
		StringBuffer logBuffer = new StringBuffer("收账过程日志\n");
		List<String> ips = ServerGroupContext.I.getGroupServerIps(groupNum);
		for(int i=0;i<ips.size();i++){
			String ip = ips.get(i);
			if(runStatusMap.get(ip) == null){
				CollectRunStatus consumerStatus = new CollectRunStatus();
				consumerStatus.setIp(ip);
				consumerStatus.setRunStatus(false);
				consumerStatus.setThreadRunTime(DispatcherConstants.COLLECTOR_TOTALTIMEOUT);
				runStatusMap.put(ip, consumerStatus);
			}
			logBuffer.append(runStatusMap.get(ip).toString()).append("\n");
		}
		
		logger.info(logBuffer.toString());
	}
	
	/**
	 * 获取httpClient
	 * @return
	 */
	protected HttpClient getHttpClient(){
		
		DefaultHttpClient httpclient = new DefaultHttpClient();
		
		httpclient.addRequestInterceptor(new HttpRequestInterceptor() {
            public void process(final HttpRequest request,final HttpContext context) throws HttpException, IOException {
                if (!request.containsHeader("Accept-Encoding")) {
                    request.addHeader("Accept-Encoding", "gzip");
                }
            }
        });

        httpclient.addResponseInterceptor(new HttpResponseInterceptor() {
            public void process(final HttpResponse response,final HttpContext context) throws HttpException, IOException {
                HttpEntity entity = response.getEntity();
                if (entity != null) {
                    Header ceheader = entity.getContentEncoding();
                    if (ceheader != null) {
                        HeaderElement[] codecs = ceheader.getElements();
                        for (int i = 0; i < codecs.length; i++) {
                            if (codecs[i].getName().equalsIgnoreCase("gzip")) {
                                response.setEntity(new GzipDecompressingEntity(response.getEntity()));
                                return;
                            }
                        }
                    }
                }
            }
        });
        
        httpclient.getParams().setParameter(CoreConnectionPNames.CONNECTION_TIMEOUT,DispatcherConstants.COLLECTOR_CONNTIMEOUT);
        httpclient.getParams().setParameter(CoreConnectionPNames.SO_TIMEOUT,DispatcherConstants.COLLECTOR_READTIMEOUT);
		
		return httpclient;
	}
	
	protected HttpGet getHttpGet(String ip,int port){
		StringBuffer buffer = new StringBuffer();
		buffer.append("http://").append(ip);
		buffer.append(":").append(port);
		buffer.append(DispatcherConstants.COLLECTOR_URL);
		return new HttpGet(buffer.toString());
	}
	
	@Override
	public boolean isCollectCompleted() {
		return collectCompleted.get();
	}

}