chinaMobile.rar_chinamobile_3704_平台_数据分析平台

package com.cm.service; import java.util.Collections; import java.util.Comparator; import java.util.Iterator; import java.util.List; import java.util.Map; import org.apache.camel.CamelContext; import org.apache.camel.Exchange; import org.apache.camel.Processor; import org.apache.camel.builder.RouteBuilder; import org.apache.camel.model.OnExceptionDefinition; import org.apache.camel.model.RouteDefinition; import org.apache.commons.collections.MapUtils; import org.apache.commons.lang.StringUtils; import org.eclipse.jetty.util.StringUtil; import org.springframework.beans.BeanUtils; import org.springframework.beans.BeansException; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.beans.factory.support.DefaultListableBeanFactory; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.stereotype.Service; import com.cm.bean.TDEndpoints; import com.cm.dao.TDCustomrouterulesMapper; import com.cm.dao.TDDatasourceMapper; import com.cm.dao.TDEndpointparamsMapper; import com.cm.dao.TDEndpointsMapper; import com.cm.dao.TDEndpointtypeMapper; import com.cm.dao.TDEndpointtypeparamsMapper; import com.cm.dao.TDExceptioninfoMapper; import com.cm.dao.TDExceptionrulesMapper; import com.cm.dao.TDJobserverMapper; import com.cm.dao.TDRoutedefinitionMapper; import com.cm.dao.TDRouteexceptionrulesMapper; import com.cm.dao.TDRoutenodeMapper; import com.cm.dao.TDRouterulesMapper; import com.cm.dto.ExceptionRule; import com.cm.dto.MyEndPoint; import com.cm.dto.RouteDef; import com.cm.dto.RouteNode; import com.cm.util.SpringContextUtil; @Service(value = "CamelEditImpl") public class CamelEditImpl implements CamelEditInterface { /** * 客户自定义规则 */ private TDCustomrouterulesMapper tDCustomrouterulesMapper; /** * 数据源 */ private TDDatasourceMapper tDDatasourceMapper; /** * 控制端点参数 */ private TDEndpointparamsMapper tDEndpointparamsMapper; /** * 控制端点 */ private TDEndpointsMapper tDEndpointsMapper; /** * 控制端点类型 */ private TDEndpointtypeMapper tDEndpointtypeMapper; /** * 控制端点类型参数 */ private TDEndpointtypeparamsMapper tDEndpointtypeparamsMapper; /** * 异常信息 */ private TDExceptioninfoMapper tDExceptioninfoMapper; /** * 异常规则 */ private TDExceptionrulesMapper tDExceptionrulesMapper; /** * 定时任务 */ private TDJobserverMapper tDJobserverMapper; /** * 路由定义 */ private TDRoutedefinitionMapper tDRoutedefinitionMapper; /** * 路由异常规则 */ private TDRouteexceptionrulesMapper tDRouteexceptionrulesMapper; /** * 路由节点 */ private TDRoutenodeMapper tDRoutenodeMapper; /** * 路由规则 */ private TDRouterulesMapper tDRouterulesMapper; @Autowired @Qualifier("camel") CamelContext camelContext; /* * 路由定义 路由控制端点 路由节点 (non-Javadoc) * * @see com.cm.service.CamelEditInterface#addCamelRoute(com.cm.dto.RouteDef, * com.cm.dto.MyEndPoint, java.util.List) */ public void addCamelRoute(final RouteDef routeDef, final MyEndPoint myEndPoint, final List<RouteNode> routeNodes) { // 第一步 增加数据库数据保存，伪代码，需后续完善 TDEndpoints tDEndpoints = new TDEndpoints(); BeanUtils.copyProperties(myEndPoint, tDEndpoints); this.tDEndpointsMapper.insert(tDEndpoints); // 第二部 开始操作路由节点 try { camelContext.addRoutes(new RouteBuilder() { @Override public void configure() throws Exception { // 初始化异常处理 errorHandler(defaultErrorHandler().maximumRedeliveries(3).redeliveryDelay(2000) .onRedelivery(new Processor() { public void process(Exchange exchange) throws Exception { System.out.println("onRedelivery"); } })); for (ExceptionRule exceptionRule : routeDef.getExceptionRules()) { OnExceptionDefinition onExceptionDefinition = onException( exceptionRule.getException().getClass()) .redeliveryDelay(exceptionRule.getRedeliverydelay()) .maximumRedeliveries(exceptionRule.getMaxredeliveries()); } ; // 初始化from节点 final RouteNode fromNode = routeNodes.get(0); if (!fromNode.getNodetype().trim().equals("0")) { throw new RuntimeException("起始节点必须为接入终端。。。"); } RouteDefinition rd = null; if (myEndPoint.istrigger()) {// 使用触发器 rd = from("quartz://" + myEndPoint.getEndpointTypeName() + "?" + "cron=" + myEndPoint.getJobtrigger().getCronexpress()).to("id:" + fromNode.getEndpointid()) .routeId(myEndPoint.getEndpointid() + ""); } else { rd = from("id:" + fromNode.getEndpointid()).routeId(myEndPoint.getEndpointid() + ""); } // 加载路由规则，按照RANK排序 List<Map<String, Object>> rules = routeDef.getRouteRule(); Collections.sort(rules, new Comparator<Map<String, Object>>() { public int compare(Map<String, Object> o1, Map<String, Object> o2) { return MapUtils.getIntValue(o1, "RANK", 1) - MapUtils.getIntValue(o2, "RANK", 1); } }); Iterator<Map<String, Object>> iterator = rules.iterator(); RouteNode node = null; Map<String, Object> rule = null; while (iterator.hasNext()) { rule = iterator.next(); // 含有自定义路由规则 if (MapUtils.getLong(rule, "CUSTOMROUTERULEID", -1L) != -1L) { break; } else { for(RouteNode routeNode : routeNodes){ if(routeNode.getRoutenodeid() !=0){ if(rule.get("ROUTENODEID").equals((routeNode.getRoutenodeid()+""))){ node = routeNode; } } } } String nodeType = node.getNodetype().trim(); ConfigurableApplicationContext configurableApplicationContext = (ConfigurableApplicationContext) SpringContextUtil .getApplicationContext(); final DefaultListableBeanFactory defaultListableBeanFactory = (DefaultListableBeanFactory) configurableApplicationContext .getBeanFactory(); // 终端节点 if ("0".equals(nodeType)) { rd = rd.to("id:" + node.getEndpointid()); } else if ("1".equals(nodeType)) {// processor节点 Processor processor = null; if (!StringUtils.isEmpty(node.getSpringnoderef())) { processor = defaultListableBeanFactory.getBean(node.getSpringnoderef(), Processor.class); } else { Class clazz = Thread.currentThread().getContextClassLoader() .loadClass(node.getNodeclasspath()); processor = (Processor) clazz.newInstance(); } rd = rd.process(processor); } else if ("3".equals(nodeType)) {// bean节点 Object o = defaultListableBeanFactory.getBean(node.getSpringnoderef()); rd = rd.bean(o); } } } }); } catch (Exception e) { e.printStackTrace(); } } /** * @param routeId * @Description:删除指定路由 */ public void deleteRoute(String routeId) { try { camelContext.removeRoute(routeId); } catch (Exception e) { e.printStackTrace(); } } }