数据发送接口源码示例

        由于项目中经常需要写各种对接的接口，此次决定花点心思设计一下，避免每次都重复开发。场景描述：重数据库中读取数据发送至WebService接口，需要对发送的数据进行确认。以往开发的痛点在于:

要操作的表及表的字段不确定
对接的WebService接口及规范不统一，此原因在于对接的第三方。
        设计思路:

抽象容器类，负责管理所有的作业，并调度Work,BaseContainer为基本实现，装载Work基本信息
public abstract class Container {
public static Logger Log = Logger.getLogger(Container.class);
public List<Work> works = new ArrayList<Work>();
public ClassPathXmlApplicationContext context;

public abstract void init();

public void start() {
Log.debug("Works Start");
for (Work work : works) {
work.start();
}
}

public void shutdown() {
Log.debug("Works Shutdown");
for (Work work : works) {
work.shutdown();
}
if (context != null) {
context.close();
}
}
}

public class BaseContainer extends Container {

public BaseContainer(ConfigurableApplicationContext context) {
this.context = (ClassPathXmlApplicationContext) context;
}

@Override
public void init() {
Log.debug("init");
Config config = context.getBean(Config.class);
List<Map<String, String>> workList = config.getWorkList();
Work work = null;
WorkParam params = null;
for (Map<String, String> workConf : workList) {
work = context.getBean("baseWork",BaseWork.class);
params = new WorkParam();
params.setRowKey(workConf.get(WorkParam.ROW_KEY));
params.setWorkName(workConf.get(WorkParam.WORK_NAME));
params.setSqlName(workConf.get(WorkParam.SQL_NAME));
params.setRequestUrl(workConf.get(WorkParam.REQUEST_URL));
params.setTemplateName(workConf.get(WorkParam.TEMPLATE_NAME));
params.setBatchSize(config.getBatchSize());
//装载模型数据
params.setTemplateData(loadModel(params.getTemplateName()));
if (workConf.get("batchSize") != null) {
params.setBatchSize(Integer.parseInt(workConf.get(WorkParam.BATCH_SIZE)));
}

work.setWorkParam(params);
work.setConfig(config);
works.add(work);
Log.info(works);
}
}

public String loadModel(String fileName) {
String result = "";
try {
result = IOUtils.toString(this.getClass().getResourceAsStream("/conf/" + fileName + ".xml"));
} catch (Exception e) {
e.printStackTrace();
}
return result;
}
}

抽象出Work，一个Work代表一个对接进程，Work拥有当前进程发送数据需要的SQL文件实际为MyBatis映射文件，有对接WebService的数据发送模版，此处实际为FreeMaker模版文件。
@Service
@Scope("prototype")
public abstract class Work extends Thread {
public static Logger Log = Logger.getLogger(Work.class);
public static ScheduledThreadPoolExecutor executer;
private CountDownLatch mDownLatch;
private boolean start = true;

protected Config config;
protected WorkParam workParam;
protected RemoteClient client;

public void init() {
if (executer == null) {
synchronized (this) {
if (executer == null) {
Log.debug("init");
executer = new ScheduledThreadPoolExecutor(config.getCorePoolSize());
}
}
}
}

public void shutdown() {
start = false;
executer.shutdown();
}

@Override
public void run() {
// must init
init();
while (start) {
List<Map<String, String>> data = loadData();
if (data.size() == 0) {
try {
Thread.sleep(config.getSleepTime());
Log.info("sleeping:" + config.getSleepTime());
} catch (InterruptedException e) {
e.printStackTrace();
}
} else {
Log.debug("size:" + data.size());
mDownLatch = new CountDownLatch(data.size());
for (Map<String, String> map : data) {
executer.submit(new Task(map));
}
try {
//1分钟超时
mDownLatch.await(1,TimeUnit.MINUTES);
} catch (InterruptedException e) {
e.printStackTrace();
}
}

}
}
public abstract List<Map<String, String>> loadData();

public abstract void sendData(Map<String, String> map);

public class Task implements Runnable {
private Map<String, String> map;

public Task(Map<String, String> map) {
this.map = map;
}

public void run() {
try {
sendData(map);
} catch (Exception e) {
e.printStackTrace();
} finally {
mDownLatch.countDown();
}
}
}

public Config getConfig() {
return config;
}

public void setConfig(Config config) {
this.config = config;
}

public WorkParam getWorkParam() {
return workParam;
}

public void setWorkParam(WorkParam workParam) {
this.workParam = workParam;
}

public RemoteClient getClient() {
return client;
}

public void setClient(RemoteClient client) {
this.client = client;
}

}

@Service
@Scope("prototype")
public class BaseWork extends Work {
@Autowired
@Qualifier("sqlSessionFactoryBean")
private SqlSessionFactory sessionFactory;

@Override
public List<Map<String, String>> loadData() {
SqlSession session = sessionFactory.openSession();
List<Map<String, String>> data = session.selectList(workParam.getSqlName() + ".select",
workParam.getBatchSize());
session.close();
return data;
}

@Override
public void sendData(Map<String, String> data) {
Log.info(data);
String key = data.get(workParam.getRowKey());
try {
client = new HttpClientToWebService(workParam.getWorkName());
String modelBean = client.makeObjectBean(workParam.getTemplateData(), data);
String result = client.call(workParam.getRequestUrl(), modelBean);
Map<String, String> resultMap = handlerResult(result);
updateUploadFlag(workParam.getSqlName(), key, resultMap);
} catch (Exception e) {
updateUploadFlag(workParam.getSqlName(), key, null);
e.printStackTrace();
Log.error("请检查数据:"+data);
}
}

public Map<String, String> handlerResult(String result) {
Log.debug(result);
Map<String, String> map = null;
Document document = null;
try {
map = new HashMap<String, String>();
document = DocumentHelper.parseText(result);
Node node = document.selectSingleNode("//inPeccancyInfoReturn[1]");
if (node != null) {
document = DocumentHelper.parseText(node.getText());
String code = document.selectSingleNode("//code[1]").getText();
String message = document.selectSingleNode("//message[1]").getText();
map.put("code", code);
map.put("message", message);
}
} catch (Exception e) {
e.printStackTrace();
}
return map;
}

public void updateUploadFlag(String region, String key, Map<String, String> result) {
Log.info("key:"+key+",result:"+result);
SqlSession session = sessionFactory.openSession();
if (result != null) {
if (Config.OK.equals(result.get("code"))) {
session.update(workParam.getSqlName() + ".success", key);
session.close();
return;
}
}
session.update(workParam.getSqlName() + ".fail", key);
session.close();
}

@Override
public String toString() {
return "BaseWork [sessionFactory=" + sessionFactory + ", workParam=" + workParam + "]";
}

}

数据发送接口定义
public interface RemoteClient {

/**
* 发送数据
*/
String call(String url, String messbean) throws Exception;

/**
* 生成模版数据
*/
String makeObjectBean(String model, Map<String, String> data) throws Exception;

}

public class HttpClientToWebService implements RemoteClient {
private String name;

public HttpClientToWebService(String name) {
this.name = name;
}

public String call(String url, String messbean) throws Exception {
String result = null;
HttpPost post = new HttpPost(url);
post.setHead
4000
er("Content-Type", "text/xml; charset=UTF-8");
post.setHeader("SOAPAction", "");
HttpEntity entity = new StringEntity(messbean, "UTF-8");
post.setEntity(entity);
HttpResponse response = HttpUtil.httpClient.execute(post);
if (response != null) {
result = IOUtils.toString(response.getEntity().getContent());
}
post.releaseConnection();
return result;
}

public String makeObjectBean(String model, Map<String, String> map) throws Exception {
String result = null;
Template template = null;
Configuration cfg = new Configuration(Configuration.VERSION_2_3_23);
cfg.setDefaultEncoding("UTF-8");
StringTemplateLoader stringTL = new StringTemplateLoader();
stringTL.putTemplate(name, model.toString());
cfg.setTemplateLoader(stringTL);
template = cfg.getTemplate(name);
ByteArrayOutputStream out = new ByteArrayOutputStream();
template.process(map, new OutputStreamWriter(out));
result = IOUtils.toString(out.toByteArray(), "UTF-8");
out.close();
return result;
}

public String getName() {
return name;
}

public void setName(String name) {
this.name = name;
}
}

工具类，HttpUtl,Lock定义
public class HttpUtil {
public static HttpClient httpClient;
static {
PoolingHttpClientConnectionManager connManager = new PoolingHttpClientConnectionManager();
// Increase max total connection to 200
connManager.setMaxTotal(200);
// Increase default max connection per route to 20
connManager.setDefaultMaxPerRoute(20);
// 请求重试处理
HttpRequestRetryHandler retryHandler = new HttpRequestRetryHandler() {
public boolean retryRequest(IOException exception, int executionCount, HttpContext context) {
if (executionCount >= 3) {// 重试三次放弃
return false;
}
if (exception instanceof NoHttpResponseException) {// 如果服务器丢掉了连接，那么就重试
return true;
}
if (exception instanceof SSLHandshakeException) {// 不要重试SSL握手异常
return false;
}
if (exception instanceof InterruptedIOException) {// 超时
return false;
}
if (exception instanceof UnknownHostException) {// 目标服务器不可达
return false;
}
if (exception instanceof ConnectTimeoutException) {// 连接被拒绝
return false;
}
if (exception instanceof SSLException) {// SSL握手异常
return false;
}
HttpClientContext clientContext = HttpClientContext.adapt(context);
HttpRequest request = clientContext.getRequest();
// 如果请求是幂等的，就再次尝试
if (!(request instanceof HttpEntityEnclosingRequest)) {
return true;
}
return false;
}
};
HttpClientBuilder builder = HttpClientBuilder.create();
builder.setConnectionManager(connManager);
builder.setRetryHandler(retryHandler);
httpClient = builder.build();
}
}

public class Lock {
@SuppressWarnings("resource")
public static synchronized boolean isLocking(String fileName) {
File file = new File("conf/" + fileName);
try {
if (!file.exists()) {
file.createNewFile();
}
final FileLock lock = new FileOutputStream(file).getChannel().tryLock();
System.out.println(lock);
if (lock == null) {
return true;
}
Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
@Override
public void run() {
try {
lock.release();
} catch (IOException e) {
e.printStackTrace();
}
}
}));
} catch (Exception e) {
e.printStackTrace();
}
return false;
}
}

Main，启动流程
public class ViolationMain {
public static ClassPathXmlApplicationContext context = null;
public static Container container = null;
public static Logger Log = Logger.getLogger(ViolationMain.class);

public static void main(String[] args) {
if (Lock.isLocking("lock.lock")) {
System.out.println();
throw new RuntimeException("One instance arready running");
}
PropertyConfigurator.configure("conf/log4j.properties");
Log.debug("System Init");
context = new ClassPathXmlApplicationContext("conf/spring.xml");
container = new BaseContainer(context);
container.init();
container.start();
Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
@Override
public void run() {
container.shutdown();
context.close();
}
}));
}
}
Spring 配置文件
spring.xml
<?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:task="http://www.springframework.org/schema/task"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:util="http://www.springframework.org/schema/util"
xsi:schemaLocation="http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-4.3.xsd 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/util http://www.springframework.org/schema/util/spring-util-4.3.xsd"> <import resource="works.xml" />
<context:property-placeholder location="conf/*.properties" />
<context:component-scan base-package="com.ehl.tvc.violation"></context:component-scan>
<bean id="config" class="com.ehl.tvc.violation.config.Config">
<property name="batchSize" value="${batchSize}"></property>
<property name="corePoolSize" value="${corePoolSize}"></property>
<property name="sleepTime" value="${sleepTime}"></property>
<property name="workList" ref="workList"></property>
</bean>
<bean id="dataSource" class="com.mchange.v2.c3p0.ComboPooledDataSource">
<property name="user" value="${db.user}"></property>
<property name="password" value="${db.password}"></property>
<property name="driverClass" value="${db.driverClass}"></property>
<property name="jdbcUrl" value="${db.jdbcUrl}"></property>
</bean>

<bean id="sqlSessionFactoryBean" class="org.mybatis.spring.SqlSessionFactoryBean">
<property name="dataSource" ref="dataSource"></property>
<property name="mapperLocations" value="conf/sql/*.xml"></property>
</bean>
</beans>


works.xml
<?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:util="http://www.springframework.org/schema/util"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util-4.3.xsd"> <util:list id="workList" list-class="java.util.ArrayList"
value-type="java.util.Map">
<map key-type="java.lang.String" value-type="java.lang.String">
<description>违规变道</description>
<entry key="rowKey" value="XH"></entry>
<entry key="workName" value="wgbd"></entry>
<entry key="templateName" value="template/wgbd" />
<entry key="sqlName" value="wgbd" />
<entry key="requestUrl"
value="http://localhost:8888/EHL_TVPDB_WEBSERVICE/services/TvpService?wsdl" />
</map>
<map key-type="java.lang.String" value-type="java.lang.String">
<description>区间超速</description>
<entry key="rowKey" value="XH"></entry>
<entry key="workName" value="qccs"></entry>
<entry key="templateName" value="template/qjcs" />
<entry key="sqlName" value="qjcs" />
<entry key="requestUrl"
value="http://localhost:8888/EHL_TVPDB_WEBSERVICE/services/TvpService?wsdl" />
</map>
<map key-type="java.lang.String" value-type="java.lang.String">
<description>车辆限行</description>
<entry key="rowKey" value="XH"></entry>
<entry key="workName" value="clxx"></entry>
<entry key="templateName" value="template/clxx" />
<entry key="sqlName" value="clxx" />
<entry key="requestUrl"
value="http://localhost:8888/EHL_TVPDB_WEBSERVICE/services/TvpService" />
</map>
</util:list>
</beans>
FreeMaker 模版文件示例
<soapenv:Envelope xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:xsd="http://www.w3.org/2001/XMLSchema" xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/"
xmlns:ser="http://service.webservice.tvp.ehl.com">
<soapenv:Header />
<soapenv:Body>
<ser:inPeccancyInfo soapenv:encodingStyle="http://schemas.xmlsoap.org/soap/encoding/">
<tvpXml xsi:type="xsd:string"><![CDATA[<?xml version="1.0" encoding="UTF-8"?>
<Data>
<DataType>Violation</DataType>
<DataBody>
<HPZL>0${HPZL}</HPZL>
<HPHM>${HPHM}</HPHM>
<WFXW>11</WFXW>
<WFSJ>${WFSJ}</WFSJ>
<WFDD>10010000000A</WFDD>
<CDBH>${CDBH}</CDBH>
<FXBH></FXBH>
<CLSD>00</CLSD>
<CSBL>00</CSBL>
<SFQJ>00</SFQJ>
<SJLY>00</SJLY>
<CJJG></CJJG>
<SBBH></SBBH>
<CJYH>tvc</CJYH>
<CJSJ>${WFSJ}</CJSJ>
<ZJWJ1>${ZJWJ1}</ZJWJ1>
<ZJWJ2></ZJWJ2>
<ZJWJ3></ZJWJ3>
<ZJWJ4></ZJWJ4>
</DataBody>
</Data>]]>
</tvpXml>
</ser:inPeccancyInfo>
</soapenv:Body>
</soapenv:Envelope>