提交ipu-database和msgframe结合使用示例

ipu-ddmp-data/.classpath

@@ -0,0 +1,36 @@
<?xml version="1.0" encoding="UTF-8"?>
<classpath>
	<classpathentry kind="src" output="target/classes" path="src/main/java">
		<attributes>
			<attribute name="optional" value="true"/>
			<attribute name="maven.pomderived" value="true"/>
		</attributes>
	</classpathentry>
	<classpathentry excluding="**" kind="src" output="target/classes" path="src/main/resources">
		<attributes>
			<attribute name="maven.pomderived" value="true"/>
		</attributes>
	</classpathentry>
	<classpathentry kind="src" output="target/test-classes" path="src/test/java">
		<attributes>
			<attribute name="optional" value="true"/>
			<attribute name="maven.pomderived" value="true"/>
		</attributes>
	</classpathentry>
	<classpathentry excluding="**" kind="src" output="target/test-classes" path="src/test/resources">
		<attributes>
			<attribute name="maven.pomderived" value="true"/>
		</attributes>
	</classpathentry>
	<classpathentry kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER/org.eclipse.jdt.internal.debug.ui.launcher.StandardVMType/jdk1.8.0_131">
		<attributes>
			<attribute name="maven.pomderived" value="true"/>
		</attributes>
	</classpathentry>
	<classpathentry kind="con" path="org.eclipse.m2e.MAVEN2_CLASSPATH_CONTAINER">
		<attributes>
			<attribute name="maven.pomderived" value="true"/>
		</attributes>
	</classpathentry>
	<classpathentry kind="output" path="target/classes"/>
</classpath>

ipu-ddmp-data/.project

@@ -0,0 +1,23 @@
<?xml version="1.0" encoding="UTF-8"?>
<projectDescription>
	<name>ipu-ddmp-data</name>
	<comment></comment>
	<projects>
	</projects>
	<buildSpec>
		<buildCommand>
			<name>org.eclipse.jdt.core.javabuilder</name>
			<arguments>
			</arguments>
		</buildCommand>
		<buildCommand>
			<name>org.eclipse.m2e.core.maven2Builder</name>
			<arguments>
			</arguments>
		</buildCommand>
	</buildSpec>
	<natures>
		<nature>org.eclipse.jdt.core.javanature</nature>
		<nature>org.eclipse.m2e.core.maven2Nature</nature>
	</natures>
</projectDescription>

ipu-ddmp-data/pom.xml

@@ -0,0 +1,154 @@
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
	<modelVersion>4.0.0</modelVersion>
	<parent>
		<groupId>com.ai.ipu</groupId>
		<artifactId>ipu-spring-libs</artifactId>
		<version>3.0</version>
	</parent>
	<groupId>com.ai.ipu</groupId>
	<artifactId>ipu-ddmp-data</artifactId>
	<version>0.0.1-SNAPSHOT</version>
	<name>ipu-ddmp-data</name>
	<repositories>
		<repository>
			<id>ipu</id>
			<name>ipu repository</name>
			<url>http://114.215.100.48:9090/nexus/content/groups/public/</url>
			<releases>
				<enabled>true</enabled>
			</releases>
			<snapshots>
				<enabled>true</enabled>
				<updatePolicy>always</updatePolicy>
			</snapshots>
		</repository>
	</repositories>
	<properties>
		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
		<msgframe>1.9.2</msgframe>
	</properties>
	<dependencies>
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-web</artifactId>
			<exclusions>
				<exclusion>
					<groupId>org.hibernate</groupId>
					<artifactId>hibernate-validator</artifactId>
				</exclusion>
				<exclusion>
					<groupId>com.fasterxml.jackson.core</groupId>
					<artifactId>jackson-databind</artifactId>
				</exclusion>
			</exclusions>
		</dependency>
		<!-- 缺省用tomcat容器 -->
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-tomcat</artifactId>
			<scope>provided</scope>
		</dependency>
		<!-- =============公共必选依赖============= -->
		<dependency>
			<groupId>javax.servlet</groupId>
			<artifactId>servlet-api</artifactId>
			<scope>provided</scope>
		</dependency>
		<dependency>
			<groupId>org.apache.commons</groupId>
			<artifactId>commons-lang3</artifactId>
		</dependency>
		<dependency>
			<groupId>junit</groupId>
			<artifactId>junit</artifactId>
			<scope>test</scope>
			<version>4.12</version>
		</dependency>
		<dependency>
			<groupId>log4j</groupId>
			<artifactId>log4j</artifactId>
		</dependency>
		<dependency>
			<groupId>dom4j</groupId>
			<artifactId>dom4j</artifactId>
		</dependency>
		<!-- msgframe相关依赖 -->
		<dependency>
			<groupId>com.ai.aif.msgframe</groupId>
			<artifactId>msgframe-common</artifactId>
			<version>${msgframe}</version>
		</dependency>
		<dependency>
			<groupId>com.ai.aif.msgframe</groupId>
			<artifactId>msgframe-client</artifactId>
			<version>${msgframe}</version>
		</dependency>
		<dependency>
			<groupId>com.ai.aif.msgframe</groupId>
			<artifactId>msgframe-server</artifactId>
			<version>${msgframe}</version>
		</dependency>
		<dependency>
			<groupId>com.ai.aif.msgframe</groupId>
			<artifactId>xmlmsgframe</artifactId>
			<version>2.6.4</version>
		</dependency>
		<dependency>
			<groupId>org.apache.xmlbeans</groupId>
			<artifactId>xmlbeans</artifactId>
			<version>2.6.0</version>
		</dependency>
		<dependency>
			<groupId>com.rabbitmq</groupId>
			<artifactId>amqp-client</artifactId>
			<version>3.6.5</version>
		</dependency>
		<dependency>
			<groupId>com.alibaba</groupId>
			<artifactId>fastjson</artifactId>
			<version>1.1.46</version>
		</dependency>
		<dependency>
			<groupId>org.hamcrest</groupId>
			<artifactId>hamcrest-core</artifactId>
			<version>1.3</version>
		</dependency>
		<dependency>
			<groupId>javax.jms</groupId>
			<artifactId>jms</artifactId>
			<version>1.1</version>
		</dependency>
		<dependency>
			<groupId>org.apache.kafka</groupId>
			<artifactId>kafka-clients</artifactId>
			<version>0.10.2.0</version>
		</dependency>
		<dependency>
			<groupId>com.ai.ipu</groupId>
			<artifactId>ipu-database</artifactId>
			<version>3.0</version>
		</dependency>
		<dependency>
			<groupId>org.aspectj</groupId>
			<artifactId>aspectjrt</artifactId>
			<version>1.8.9</version>
		</dependency>
		<!-- https://mvnrepository.com/artifact/org.aspectj/aspectjtools -->
		<dependency>
			<groupId>org.aspectj</groupId>
			<artifactId>aspectjtools</artifactId>
			<version>1.8.9</version>
		</dependency>
		<dependency>
			<groupId>org.aspectj</groupId>
			<artifactId>aspectjweaver</artifactId>
			<version>1.7.4</version>
		</dependency>
	</dependencies>
</project>

ipu-ddmp-data/src/main/java/com/ai/ipu/ddmp/data/MainModule.java

@@ -0,0 +1,37 @@
package com.ai.ipu.ddmp.data;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.web.servlet.ServletComponentScan;
import com.ai.aif.msgframe.consumer.MfConsumerClient;
import com.ai.aif.msgframe.consumer.MfServiceStartup;
import com.ai.ipu.ddmp.data.config.DataConfig;
import com.ai.ipu.ddmp.data.service.DataManagerService;
import com.ai.ipu.ddmp.data.util.SpringUtil;
@SpringBootApplication
@ServletComponentScan
public class MainModule {
	@Autowired
    public DataConfig dataConfig;
	@Autowired
	public DataManagerService dataManagerService;
	
	public static void main(String[] args) {
		// TODO Auto-generated method stub
		SpringApplication.run(MainModule.class, args);
		
		//启动msgFrame消费者
		MfServiceStartup.main();
//		String[] topics = "tab_ipu_oper|tab_ipu_plugin|tab_ipu_page|tab_ipu_data|tab_ipu_crash|tab_ipu_device".split("\\|");
//		for (String topic:topics)
//		{
//			MfConsumerClient.subscribe(topic, "*",
//				"com.ai.ipu.ddmp.data.consumer.ConsumerProcessorImpl");
//		}
	}
}

ipu-ddmp-data/src/main/java/com/ai/ipu/ddmp/data/aop/TranscationManager.java

@@ -0,0 +1,43 @@
package com.ai.ipu.ddmp.data.aop;
import org.aspectj.lang.JoinPoint;
import org.aspectj.lang.annotation.After;
import org.aspectj.lang.annotation.AfterThrowing;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.annotation.Before;
import org.springframework.stereotype.Component;
import com.ai.ipu.basic.log.ILogger;
import com.ai.ipu.basic.log.IpuLoggerFactory;
import com.ai.ipu.database.conn.SqlSessionManager;
@Aspect //声明该类是切面类
@Component//配置文件中启动自动扫描功能
public class TranscationManager {
	protected final transient ILogger log = IpuLoggerFactory.createLogger(TranscationManager.class);
	
	@Before("execution(* com.ai.ipu.ddmp.data.service.*.*(..))")
	public void startTranscation(){
		log.debug("开始数据库操作");
	}
	@AfterThrowing(pointcut = "execution(* com.ai.ipu.ddmp.data.service.*.*(..))", throwing = "e")
    private void doAfterThrow(JoinPoint joinPoint,  Throwable e) {
		e.printStackTrace();
		long start = System.currentTimeMillis();
		SqlSessionManager.rollbackAll();
		log.debug("回退数据库连接耗时：" + (System.currentTimeMillis() - start));
		
//		log.debug("数据库连接全量回收");
//		SqlSessionManager.closeAll();
	}
	@After("execution(* com.ai.ipu.ddmp.data.service.*.*(..))")
	public void commitTranscation(){
		long start = System.currentTimeMillis();
		SqlSessionManager.commitAll();
		log.debug("提交数据库连接耗时：" + (System.currentTimeMillis() - start));
		
	}
	
}

ipu-ddmp-data/src/main/java/com/ai/ipu/ddmp/data/config/DataConfig.java

@@ -0,0 +1,55 @@
package com.ai.ipu.ddmp.data.config;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;
/**
 * DataConfig
 * 
 * @author：weihf@asiainfo.com
 * @2018-06-28
 * @since 1.0
 */
@Component
@ConfigurationProperties(prefix = "data.config")
public class DataConfig {
	private int batchSize;
    private int lingerMs;
    private String topics;
    private String tables;
    private String seperator;
    
	public int getBatchSize() {
		return batchSize;
	}
	public void setBatchSize(int batchSize) {
		this.batchSize = batchSize;
	}
	public int getLingerMs() {
		return lingerMs;
	}
	public void setLingerMs(int lingerMs) {
		this.lingerMs = lingerMs;
	}
	public String[] getTopics() {
		return topics.split(seperator==null?"|":seperator);
	}
	public void setTopics(String topics) {
		this.topics = topics;
	}
	public String[] getTables() {
		return tables.split(seperator==null?"|":seperator);
	}
	public void setTables(String tables) {
		this.tables = tables;
	}
	public String getSeperator() {
		return seperator;
	}
	public void setSeperator(String seperator) {
		this.seperator = seperator;
	}
}

ipu-ddmp-data/src/main/java/com/ai/ipu/ddmp/data/consumer/ConsumerProcessorImpl.java

@@ -0,0 +1,25 @@
package com.ai.ipu.ddmp.data.consumer;
import com.ai.aif.msgframe.common.IConsumerProcessor;
import com.ai.aif.msgframe.common.exception.ConsumerException;
import com.ai.aif.msgframe.common.message.MsgFMessage;
import com.ai.aif.msgframe.common.message.MsgFTextMessage;
import com.ai.ipu.ddmp.data.manager.DataManager;
public class ConsumerProcessorImpl implements IConsumerProcessor {
	
	@SuppressWarnings("rawtypes")
	public Object process(MsgFMessage msg) throws ConsumerException {
	
		if (msg instanceof MsgFTextMessage) {
			System.out.println("receive message" + "   " +msg.getTopic() + msg.getMsgId() + "   " + msg.getFilterTag() + "       "
					+ ((MsgFTextMessage) msg).getText());
			DataManager.putSql(msg.getTopic(), ((MsgFTextMessage) msg).getText());
		}
		// 返回Boolean类型的false，代表消息消费失败，重新消费，其他表示消费成功
		return new Boolean(true);
	}
}

ipu-ddmp-data/src/main/java/com/ai/ipu/ddmp/data/dao/DataDao.java

@@ -0,0 +1,32 @@
package com.ai.ipu.ddmp.data.dao;
import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import com.ai.ipu.database.dao.impl.AbstractBizDao;
public class DataDao extends AbstractBizDao {
	public DataDao(String connName) throws IOException {
		super(connName);
	}
	public int save(String sql) throws Exception
	{
		return dao.executeInsert(sql);
	}
	
	public int save(String sql, Map<String, Object> param) throws Exception
	{
		return dao.executeInsert(sql, param);
	}
	
	public List<Map<String, Object>> takeTableCols(String tableName) throws Exception {
    	String sql = "select COLUMN_NAME, IS_NULLABLE, DATA_TYPE  from information_schema.COLUMNS where table_name =#{tableName} and EXTRA != 'auto_increment'";
    	Map<String, Object> param = new HashMap<String, Object>();
    	param.put("tableName", tableName);
    	return  dao.executeSelect(sql, param);   
    }
}

ipu-ddmp-data/src/main/java/com/ai/ipu/ddmp/data/manager/DataManager.java

@@ -0,0 +1,139 @@
package com.ai.ipu.ddmp.data.manager;
import java.util.ArrayList;
import java.util.List;
import java.util.Queue;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;
import com.ai.ipu.ddmp.data.config.DataConfig;
import com.ai.ipu.ddmp.data.service.DataManagerService;
import com.ai.ipu.ddmp.data.util.SpringUtil;
@Component
public class DataManager implements CommandLineRunner {
	private static final Logger logger = LoggerFactory.getLogger(DataManager.class);
	private static final AtomicInteger batchSize = new AtomicInteger(0);
	private static final ConcurrentMap<String /* table name*/, Queue<String /* sql*/>> sqlTable = new ConcurrentHashMap<String, Queue<String>>();
	private static final ConcurrentMap<String /* table name*/, AtomicBoolean /* lock */> tableLock = new ConcurrentHashMap<String, AtomicBoolean>();
	private static final ConcurrentMap<String /* table name*/, AtomicInteger /* size */> tableSize = new ConcurrentHashMap<String, AtomicInteger>();
	
	public void run(String... args) throws Exception {
		// TODO Auto-generated method stub
		DataConfig dataConfig = (DataConfig) SpringUtil.getBean("dataConfig");
		/*Queue<String> sqlQue = new ConcurrentLinkedQueue();
		AtomicBoolean lock = new AtomicBoolean(false);
		AtomicInteger queSize = new AtomicInteger(0);
		
		sqlQue.add("a");
		queSize.incrementAndGet();
		
		sqlQue.add("b");
		queSize.incrementAndGet();
		sqlQue.add("c");
		queSize.incrementAndGet();
		
		sqlTable.put("test", sqlQue);
		tableLock.put("test", lock);
		tableSize.put("test", queSize);*/
		
		if (dataConfig.getBatchSize()<=0)
			batchSize.set(1000);
		else
			batchSize.set(dataConfig.getBatchSize());
		
		if (dataConfig.getLingerMs() == 0)
			return;
        
		Timer timer = new Timer();
        timer.scheduleAtFixedRate(new TimerTask() {
            public void run() {
            	logger.debug("开始做定时任务");
            	
            	/* 如果到达队列定义长度或时间间隔时，通过tableLock判断是否已经在对队列做取处理。不允许同时对队列做取操作*/
            	for(String key : sqlTable.keySet())
            	{
            		if (tableSize.get(key) == null)
            			tableSize.put(key, new AtomicInteger(0));
            		if (tableLock.get(key) == null)
            			tableLock.put(key, new AtomicBoolean(false));
            		
            		tableLock.get(key).set(true);
            		List sqlList = new ArrayList();
           			String sql;
    				while ((sql = sqlTable.get(key).poll())  != null)
           			{
    					tableSize.get(key).decrementAndGet();
    					sqlList.add(sql);
           			}
    				DataManagerService service = (DataManagerService) SpringUtil.getBean("dataManagerService");
    				try {
    					service.saveBatch(key, sqlList);
					} catch (Exception e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
    				tableLock.get(key).set(false);
            	}
    			
            	logger.debug("定时任务完成");
            }
        }, 0, dataConfig.getLingerMs()*1000);        
	}
	public static boolean putSql(String topicName, String sql) {
		// TODO Auto-generated method stub
		if (sql == null || "".equals(sql.trim())) {
			return false;
		}
		if (tableSize.get(topicName) == null)
			tableSize.put(topicName, new AtomicInteger(0));
		if (tableLock.get(topicName) == null)
			tableLock.put(topicName, new AtomicBoolean(false));
		
		if (sqlTable.get(topicName) == null)
			sqlTable.put(topicName, new ConcurrentLinkedQueue());
		
		/* 如果到达队列定义长度或时间间隔时，通过tableLock判断是否已经在对队列做取处理。不允许同时对队列做取操作*/
		if (tableSize.get(topicName).get()>=batchSize.intValue() && !tableLock.get(topicName).get())
		{
			tableLock.get(topicName).set(true);
			List sqlList = new ArrayList();
   			String sqlStr;
			while ((sqlStr = sqlTable.get(topicName).poll())  != null)
   			{
				tableSize.get(topicName).decrementAndGet();
				sqlList.add(sqlStr);
   			}
			DataManagerService service = (DataManagerService) SpringUtil.getBean("dataManagerService");
			try {
				service.saveBatch(topicName, sqlList);
			} catch (Exception e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			tableLock.get(topicName).set(false);
		}
		sqlTable.get(topicName).add(sql);
		int size = tableSize.get(topicName).incrementAndGet();
		logger.debug("sqlTable size:" + size);
		return true;
	}
	
}

ipu-ddmp-data/src/main/java/com/ai/ipu/ddmp/data/service/DataManagerService.java

@@ -0,0 +1,180 @@
package com.ai.ipu.ddmp.data.service;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.springframework.stereotype.Service;
import com.ai.ipu.database.dao.IpuDaoManager;
import com.ai.ipu.ddmp.data.config.DataConfig;
import com.ai.ipu.ddmp.data.dao.DataDao;
import com.ai.ipu.ddmp.data.util.SpringUtil;
import com.alibaba.fastjson.JSONObject;
@Service
public class DataManagerService {
	public int save(String topicName, String value) throws Exception {
		// TODO Auto-generated method stub
		DataDao dao = IpuDaoManager.takeDao(DataDao.class, "test");
		String[] topics = ((DataConfig) SpringUtil.getBean("dataConfig")).getTopics();
		String[] tables = ((DataConfig) SpringUtil.getBean("dataConfig")).getTables();
		String tableName = "";
		for (int i=0;i<tables.length;i++)
		{
			if (topics[i].equals(topicName))
			{
				tableName = tables[i];
				break;
			}
		}
		List<Map<String, Object>> cols = dao.takeTableCols(tableName);
		StringBuffer sql = new StringBuffer();
		sql.append("insert into ")
		.append(tableName)
		.append(" ( ");
		
		int j=0;
		for (Map<String, Object> col:cols)
		{
			if (j>0)
				sql.append(",");
			sql.append(col.get("COLUMN_NAME"));
			j++;
		}
		
		sql.append(" ) ") 
		.append(" values ");
		
		Map<String, Object> param = new HashMap<String, Object>();
		JSONObject jsValue= JSONObject.parseObject(value);
		sql.append("(");
		int k=0;
		for (Map<String, Object> col:cols)
		{
			if (jsValue.get(col.get("COLUMN_NAME")) != null)
			{
				
				if (k>0)
					sql.append(",");
				param.put((String)col.get("COLUMN_NAME"), jsValue.get(col.get("COLUMN_NAME")));
				sql.append("#{")
				.append(col.get("COLUMN_NAME"))
				.append("}");
				k++;
			}
			else if (col.get("IS_NULLABLE").equals("Y"))
			{
				if (k>0)
					sql.append(",");
				if ("datetime".equalsIgnoreCase((String) col.get("DATA_TYPE")))
					;
				else if ("date".equalsIgnoreCase((String) col.get("DATA_TYPE")))
					;
				else if ("timestamp".equalsIgnoreCase((String) col.get("DATA_TYPE")))
					;
				else
					param.put((String)col.get("COLUMN_NAME"),"null");
				sql.append("#{")
				.append(col.get("COLUMN_NAME"))
				.append("}");
				k++;
			}
			else
				throw new Exception(col.get("COLUMN_NAME")+" is not null.");
				
		}
		sql.append(")");
		return dao.save(sql.toString(), param);
	}
	public int saveBatch(String topicName, List<String> valueList) throws Exception {
		// TODO Auto-generated method stub
		if (valueList == null || valueList.size() == 0)
			return -1;
		else if (valueList.size() == 1)
			return save(topicName, valueList.get(0));
		
		DataDao dao = IpuDaoManager.takeDao(DataDao.class, "test");
		String[] topics = ((DataConfig) SpringUtil.getBean("dataConfig")).getTopics();
		String[] tables = ((DataConfig) SpringUtil.getBean("dataConfig")).getTables();
		String tableName = "";
		for (int i=0;i<tables.length;i++)
		{
			if (topics[i].equals(topicName))
			{
				tableName = tables[i];
				break;
			}
		}
		List<Map<String, Object>> cols = dao.takeTableCols(tableName);
		StringBuffer sql = new StringBuffer();
		sql.append("insert into ")
		.append(tableName)
		.append(" ( ");
		
		int j=0;
		for (Map<String, Object> col:cols)
		{
			if (j>0)
				sql.append(",");
			sql.append(col.get("COLUMN_NAME"));
			j++;
		}
		
		sql.append(" ) ") 
		.append(" values ");
		j=0;
		for (String value:valueList)
		{
			JSONObject jsValue= JSONObject.parseObject(value);
			if (j>0)
				sql.append(",");
			sql.append("(");
			int k=0;
			for (Map<String, Object> col:cols)
			{
				if (jsValue.get(col.get("COLUMN_NAME")) != null)
				{
					
					if (k>0)
						sql.append(",");
					if ("varchar".equalsIgnoreCase((String) col.get("DATA_TYPE")) || "char".equalsIgnoreCase((String) col.get("DATA_TYPE")))
						sql.append("'").append(jsValue.get(col.get("COLUMN_NAME"))).append("'");
					else if ("datetime".equalsIgnoreCase((String) col.get("DATA_TYPE")))
						;
					else if ("date".equalsIgnoreCase((String) col.get("DATA_TYPE")))
						;
					else if ("timestamp".equalsIgnoreCase((String) col.get("DATA_TYPE")))
						;
					else
						sql.append(jsValue.get(col.get("COLUMN_NAME")));
					k++;
				}
				else if (col.get("IS_NULLABLE").equals("Y"))
				{
					if (k>0)
						sql.append(",");
					if ("datetime".equalsIgnoreCase((String) col.get("DATA_TYPE")))
						;
					else if ("date".equalsIgnoreCase((String) col.get("DATA_TYPE")))
						;
					else if ("timestamp".equalsIgnoreCase((String) col.get("DATA_TYPE")))
						;
					else
						sql.append("null");
					k++;
				}
				else
					throw new Exception(col.get("COLUMN_NAME")+" is not null.");
					
			}
			sql.append(")");
			j++;
		}
		return dao.save(sql.toString());
	}
	
}

ipu-ddmp-data/src/main/java/com/ai/ipu/ddmp/data/util/SpringUtil.java

@@ -0,0 +1,40 @@
package com.ai.ipu.ddmp.data.util;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.stereotype.Component;
@Component
public class SpringUtil implements ApplicationContextAware {
	private static ApplicationContext applicationContext = null;
	
	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
		if(SpringUtil.applicationContext == null){
            SpringUtil.applicationContext  = applicationContext;
        }
	}
	//获取applicationContext
    public static ApplicationContext getApplicationContext() {
        return applicationContext;
    }
    //通过name获取 Bean.
    public static Object getBean(String name){
        return getApplicationContext().getBean(name);
    }
    //通过class获取Bean.
    public static <T> T getBean(Class<T> clazz){
        return getApplicationContext().getBean(clazz);
    }
    //通过name,以及Clazz返回指定的Bean
    public static <T> T getBean(String name,Class<T> clazz){
        return getApplicationContext().getBean(name, clazz);
    }
}

ipu-ddmp-data/src/main/resources/application.properties

@@ -0,0 +1,30 @@
#设置服务器端口
#也可以通过启动命令行参数实现：java -jar myproject.jar --server.port=9084
server.port=9999
is.spring.boot=true
web.root=webapp/
spring.mvc.static-path-pattern=/**
spring.resources.static-locations=classpath:/${web.root}
#context-path必须以/开头
server.context-path=/ipuDdmpData
spring.application.name=ipuDdmpData
spring.application.index=ipuDdmpData
logging.file=./logs/ipuDdmpData.log
#logging.config=classpath:logback.xml
#data.config相关配置，所有配置都可以通过--启动参数覆盖
#队列长度
data.config.batchSize=10000
#时间间隔，单位：秒
data.config.lingerMs=5
#设置接收主题
#data.config.topics=ipu_oper|ipu_plugin|ipu_page|ipu_data|ipu_crash|ipu_device
data.config.topics=ipu_data
#每个主题对应的消息需要导入的数据库表。不同主题对应的表之间用|分隔
#data.config.tables=tab_ipu_oper|tab_ipu_plugin|tab_ipu_page|tab_ipu_data|tab_ipu_crash|tab_ipu_device
data.config.tables=tmp_info
#topics和tables的分隔符,由于|有特殊含义，因此需要写为\\|
data.config.seperator=\\|

ipu-ddmp-data/src/main/resources/applicationContext.xml

@@ -0,0 +1,17 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xmlns:aop="http://www.springframework.org/schema/aop"
    xmlns:context="http://www.springframework.org/schema/context"
    xsi:schemaLocation="http://www.springframework.org/schema/beans 
        http://www.springframework.org/schema/beans/spring-beans.xsd
        http://www.springframework.org/schema/context 
        http://www.springframework.org/schema/context/spring-context-4.3.xsd
        http://www.springframework.org/schema/aop 
        http://www.springframework.org/schema/aop/spring-aop-4.3.xsd" >
        <!-- 自动扫描包下的类，并将其实例化。多个包之间用,隔开 -->
        <context:component-scan base-package="com.ai.ipu.ddmp.data"></context:component-scan>
        <!-- 配置文件中启动AspectJ的注解功能 ,默认是false，要将其改为true -->
        <aop:aspectj-autoproxy proxy-target-class="true"></aop:aspectj-autoproxy> 
</beans>

ipu-ddmp-data/src/main/resources/destination_rule.xml

@@ -0,0 +1,6 @@
<?xml version="1.0" encoding="utf-8"?>
<destinationRuleCfg xmlns="http://www.asiainfo.com/msgframe/destinationrule">
	<subject name="TagTest">
		<area name="tag1" desc="tag1" number="10"/>
	</subject>
</destinationRuleCfg>	

ipu-ddmp-data/src/main/resources/ipu-mybatis-config.xml

@@ -0,0 +1,40 @@
<?xml version="1.0" encoding="UTF-8" ?>
<!DOCTYPE configuration
    PUBLIC "-//mybatis.org//DTD Config 3.0//EN"
    "http://mybatis.org/dtd/mybatis-3-config.dtd">
<configuration>
    <settings>
		<setting name="defaultFetchSize" value="1000" /> <!-- 结果集获取数量提示值，分批传输 -->
	</settings>
    <plugins>
        <!-- 分页插件，可根据参数定制化 -->
	    <plugin interceptor="com.github.pagehelper.PageInterceptor">
	        <!-- config params as the following -->
		</plugin>
	</plugins>
	<environments default="test">
		<environment id="test">
			<transactionManager type="JDBC" />
			<dataSource type="com.ai.ipu.database.datasource.C3P0DataSourceFactory">
				<property name="driverClass" value="com.mysql.jdbc.Driver" />
				<!--<property name="jdbcUrl" value="jdbc:mysql://121.42.183.206:3317/test?useSSL=false" />  -->
				<property name="jdbcUrl" value="jdbc:mysql://10.19.13.54:6601/test?useSSL=false" />
				<property name="user" value="ipu" />
				<property name="password" value="ipumysql" />
				<!-- 连接池用完时,等待获取新连接的时间 (毫秒) -->
				<property name="checkoutTimeout" value="5000" />
				<!--定义在从数据库获取新连接失败后重复尝试的次数。Default: 30 -->
				<property name="acquireRetryAttempts" value="5" />
				<!--两次连接中间隔时间，单位毫秒。Default: 1000 -->
				<property name="acquireRetryDelay" value="1000" />
				<property name="initialPoolSize" value="50" />
				<property name="minPoolSize" value="50" />
				<property name="maxPoolSize" value="50" />
				<property name="maxIdleTime" value="600" />
				<property name="idleConnectionTestPeriod" value="60" />
				<property name="preferredTestQuery" value="SELECT 1" />
			</dataSource>
		</environment>
	</environments>
</configuration>

ipu-ddmp-data/src/main/resources/logback.xml

@@ -0,0 +1,53 @@
<?xml version="1.0" encoding="UTF-8"?>
<configuration>
	<include resource="org/springframework/boot/logging/logback/base.xml" /> <!-- logback 提供的基本配置 -->
	<!-- 控制台 -->
	<appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
		<encoder charset="UTF-8">
			<!-- <pattern>[%d{yyyy-MM-dd HH:mm:ss.SSS}] %5p %m%n</pattern> -->
			<pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} {%thread} %-5level %logger{50}-%msg%n</pattern>
			<charset>UTF-8</charset> <!-- 解决中文乱码问题 -->
		</encoder>
	</appender>
	<!-- 记录到文件 (每天一个文件) -->
	<appender name="FILE"
		class="ch.qos.logback.core.rolling.RollingFileAppender">
		<file>/data/test/logs/ipuDdmpData.log</file><!-- 
			最新的log文件名 -->
		<append>true</append>
		<rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
			<!-- 历史文件名 -->
			<fileNamePattern>/data/test/logs/ipuDdmpData-%d{yyyy-MM-dd}.%i.log
			</fileNamePattern>
			<timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
				<maxFileSize>104857600</maxFileSize>
			</timeBasedFileNamingAndTriggeringPolicy>
			<MaxHistory>30</MaxHistory><!-- 保留 30 天的日志 -->
		</rollingPolicy>
		<encoder>
			<!-- <pattern>[%d{yyyy-MM-dd HH:mm:ss.SSS}] %5p %m%n</pattern> -->
			<pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} {%thread} %-5level %logger{50}-%msg%n</pattern>
			<!-- <charset class="java.nio.charset.Charset">UTF-8</charset> -->
			<charset>UTF-8</charset>  <!--解决中文乱码问题 -->
		</encoder>
	</appender>
	<!-- <root level="DEBUG"> -->
	<!-- <root level="INFO"> <appender-ref ref="CONSOLE" /> <appender-ref ref="FILE" 
		/> </root> -->
	<!-- 将上面两个 appender 关联到我们的项目 -->
	<logger name="com.ai.ipu.ddmp.data" level="DEBUG"
		additivity="false"> <!-- name 是项目包名，为了方便调试，输出 DEBUG 级别及其以上的log -->
		<appender-ref ref="CONSOLE" />
		<appender-ref ref="FILE" />
	</logger>
	<!-- 其他框架的日志输出 -->
	<property name="LOG_FILE"
		value="/data/test/logs/spring-boot-tmp.log" />
	<logger name="org.springframework" level="INFO" />   <!-- spring 包下的 logger, 只输出 INFO 级别的 -->
</configuration> 

ipu-ddmp-data/src/main/resources/msgframe-config.xml

@@ -0,0 +1,66 @@
<?xml version="1.0" encoding="UTF-8"?>
<msgframeCfg xmlns="http://www.asiainfo.com/msgframe">
	<centerCfg>
		<!--［1-1］ 中心名称  定义当前中心或者当前系统的名称编码，全局配置唯一性-->
		<name>orderCenter</name>
		<!-- 全局配置  影响全局的一些参数配置，如入职中心的开关配置，整体的序列化方式等，全局配置唯一性 -->
		<globalCfg>
			<!-- 消息生产端消息注入处理类，可以不配或者多配，在发送时可以在消息发送前执行，配置类的全路径且继承接口 com.cmos.msgframe.common.IProsInjectionProcessor -->
			<!-- 		<prodInjection>sadsa</prodInjection> -->
			<!-- 消息消费端消息注入处理类，可以不配或者多配，在消费时可以在消息处理业务前执行，配置类的全路径且继承接口 com.cmos.msgframe.common.IConsInjectionProcessor -->
			<!-- 		<consInjection>sadsa</consInjection> -->
		</globalCfg>
		<!-- 总的主题列表配置信息，包含所有的主题配置信息，全局配置唯一性 -->
		<destinations>
		    <queue name="ipu_data" belong="orderCenter"/>
			<queue name="PerfTest" belong="orderCenter"/>
			<queue name="TagTest" belong="orderCenter" ruleClass="com.ai.aif.msgframe.common.route.impl.CustomDestinationRule"/>
			<queue name="OrderTest" order="true" belong="orderCenter" number="1"/>
			<queue name="TransactionTest" belong="orderCenter"/>
		</destinations>
		<!-- 消费端的配置信息，主要包含订阅的主题和消息的业务实现类，全局配置唯一性 -->
		<subscribes>
		    <!--   主题订阅信息，订阅在<destinations></destinations>标签中的主题消息 -->
		    <subscribe subDestination="ipu_data" consumeType="push">
				<implclass>com.ai.ipu.ddmp.data.consumer.ConsumerProcessorImpl</implclass>
			</subscribe>
			<!-- tag:可选，默认是*。配合destinations->ruleClass，如果有多个过滤条件，请按照a||b -->
			<subscribe subDestination="PerfTest" consumeType="push">
				<!-- 配置实现类，利用管道流的思想，按照配置顺序依次处理消息.只支持异步消息 -->
				<implclass>com.ai.ipu.ddmp.data.consumer.ConsumerProcessorImpl</implclass>
			</subscribe>
			<subscribe subDestination="TagTest" consumeType="push">
				<implclass>com.ai.ipu.ddmp.data.consumer.ConsumerProcessorImpl</implclass>
			</subscribe>
			<subscribe subDestination="OrderTest" consumeType="push">
				<implclass>com.ai.ipu.ddmp.data.consumer.ConsumerProcessorImpl</implclass>
			</subscribe>
			<subscribe subDestination="TransactionTest" consumeType="push">
				<implclass>com.ai.ipu.ddmp.data.consumer.ConsumerProcessorImpl</implclass>
			</subscribe>
		</subscribes>
		<!-- 主要配置中心(系统)和其所属集群的关系，全局配置唯一性 -->
		<centers>
			<center name="orderCenter" containClusters="cluster-1" />
		</centers>
		<!-- 集群配置，主要配置当前中心或者系统用到的所有集群信息，全局配置唯一性 -->
		<clusters>
		<!--RocketMQ、Kafka、ActiveMQ  -->
			<cluster name="cluster-1" type="Kafka">
  				<!-- <url>10.1.243.19:9876;10.1.243.20:9876</url> -->
  				     <url>10.19.13.54:9092</url>
			</cluster>
		</clusters>
		<!-- 持久化配置区，主要包括生产端和消费端的异常持久化配置、生产端和消费端的日志持久化配置，全局配置唯一性 -->
		<persistence>
			<exceptionPersistence>
				<!-- 生产者异常处理,必须实现com.ai.aif.msgframe.common.ex.exception.IExceptionPersitence接口 -->
				<producerExceptionClass>com.ai.aif.msgframe.producer.ex.ProducerExceptionHandle</producerExceptionClass>
				<!-- 消费者异常处理,必须实现com.ai.aif.msgframe.common.ex.exception.IExceptionPersitence接口 -->
				<consumerExceptionClass>com.ai.aif.msgframe.consumer.ex.ConsumerExceptionHandle</consumerExceptionClass>
			</exceptionPersistence>
		</persistence>
	</centerCfg>
</msgframeCfg>

ipu-ddmp-data/src/test/java/SendMessageTest.java

@@ -0,0 +1,62 @@
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.ai.aif.msgframe.MfProducerClient;
import com.ai.aif.msgframe.common.message.MsgFTextMessage;
public class SendMessageTest {
	private static final Logger logger = LoggerFactory.getLogger(SendMessageTest.class);
	private static final ExecutorService pool = Executors.newFixedThreadPool(5);
	@org.junit.Test
	public void sendMessageTest() {
		try {
//			for (int i = 0; i < 10; i++) {
				pool.submit(new ProducerThreadTest());
//			}
		} catch (Exception e) {
			throw new RuntimeException(e);
		}
		
		while(true){
			
		}
	}
	class ProducerThreadTest implements Runnable {
		public void run() {
			try {
				MfProducerClient client = new MfProducerClient();
				for (int i = 0; i < 10; i++) {
					MsgFTextMessage message = new MsgFTextMessage();
					message.setText("{info_name:\"abc"+i+"\",info_class:"+i+"}");
					client.send("ipu_data", message);
					logger.info("发送成功，" + message);
					Thread.currentThread().sleep(1000);
				}
			} catch (Exception e) {
				throw new RuntimeException(e);
			}
		}
	}
}

ipu-ddmp-data/src/test/java/SendOnewayMessageTest.java

@@ -0,0 +1,62 @@
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.ai.aif.msgframe.MfProducerClient;
import com.ai.aif.msgframe.common.message.MsgFTextMessage;
public class SendOnewayMessageTest {
	
	private static final Logger logger = LoggerFactory.getLogger(SendOnewayMessageTest.class);
	private static final ExecutorService pool = Executors.newFixedThreadPool(5);
	@org.junit.Test
	public void sendMessageTest() {
		try {
			for (int i = 0; i < 10; i++) {
				pool.submit(new ProducerThreadTest());
			}
		} catch (Exception e) {
			throw new RuntimeException(e);
		}
		
		while(true){
			
		}
	}
	class ProducerThreadTest implements Runnable {
		public void run() {
			try {
				MfProducerClient client = new MfProducerClient();
				for (int i = 0; i < 10000; i++) {
					
					MsgFTextMessage message = new MsgFTextMessage();
					message.setText("oneway message test " + i);
					client.sendOneway("PerfTest", message);
					logger.info("发送成功，" + message);
					Thread.currentThread().sleep(1000);
				}
			} catch (Exception e) {
				throw new RuntimeException(e);
			}
		}
	}
}

ipu-ddmp-data/src/test/java/SendOrderMessageTest.java

@@ -0,0 +1,63 @@
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.ai.aif.msgframe.MfProducerClient;
import com.ai.aif.msgframe.common.message.MsgFTextMessage;
public class SendOrderMessageTest {
	private static final Logger logger = LoggerFactory.getLogger(SendOrderMessageTest.class);
	private static final ExecutorService pool = Executors.newFixedThreadPool(5);
	@org.junit.Test
	public void sendMessageTest() {
		try {
			for (int i = 0; i < 10; i++) {
				pool.submit(new ProducerThreadTest());
			}
		} catch (Exception e) {
			throw new RuntimeException(e);
		}
		
		while(true){
			
		}
	}
	class ProducerThreadTest implements Runnable {
		public void run() {
			try {
				MfProducerClient client = new MfProducerClient();
				String orderId = "2013";// orderId对于同一组顺序消息必须一致
				for (int i = 0; i < 3; i++) {
					MsgFTextMessage orderMessage = new MsgFTextMessage();
					orderMessage.setText("order message " + i);
					client.sendOrderMsg("OrderTest", orderMessage, orderId);
					
					logger.info("发送成功，" + orderMessage);
					Thread.currentThread().sleep(1000);
				}
			} catch (Exception e) {
				throw new RuntimeException(e);
			}
		}
	}
}

ipu-ddmp-data/src/test/java/SendTagMessageTest.java

@@ -0,0 +1,63 @@
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.ai.aif.msgframe.MfProducerClient;
import com.ai.aif.msgframe.common.message.MsgFTextMessage;
public class SendTagMessageTest {
	private static final Logger logger = LoggerFactory.getLogger(SendTagMessageTest.class);
	private static final ExecutorService pool = Executors.newFixedThreadPool(5);
	@org.junit.Test
	public void sendMessageTest() {
		try {
			for (int i = 0; i < 10; i++) {
				pool.submit(new ProducerThreadTest());
			}
		} catch (Exception e) {
			throw new RuntimeException(e);
		}
		
		while(true){
			
		}
	}
	class ProducerThreadTest implements Runnable {
		public void run() {
			try {
				MfProducerClient client = new MfProducerClient();
				
				for (int i = 0; i < 10000; i++) {
					MsgFTextMessage message = new MsgFTextMessage();
					message.setText("message tag test" + i);
					message.setFilterTag("tag1");// 设置消息tag
					client.send("TagTest", message);
					logger.info("发送成功，" + message);
					Thread.currentThread().sleep(1000);
				}
			} catch (Exception e) {
				throw new RuntimeException(e);
			}
		}
	}
}