数据同步工具类，使用Rest接口同步数据

import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.JSONObject; import com.google.gson.Gson; import com.google.gson.reflect.TypeToken; import com.mchange.v2.c3p0.ComboPooledDataSource; import org.apache.commons.dbutils.QueryRunner; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.beans.PropertyVetoException; import java.io.IOException; import java.io.UnsupportedEncodingException; import java.math.BigDecimal; import java.sql.SQLException; import java.sql.Timestamp; import java.text.ParseException; import java.text.SimpleDateFormat; import java.time.Instant; import java.time.LocalDate; import java.time.ZoneOffset; import java.time.temporal.ChronoUnit; import java.util.*; import java.util.concurrent.CompletionService; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorCompletionService; import java.util.concurrent.Future; import java.util.concurrent.atomic.AtomicBoolean; /** * Description: * * @author **/ public class SynWork implements Runnable { private Logger logger = LoggerFactory.getLogger(SynWork.class); ComboPooledDataSource cpds; String sql; AtomicBoolean inWork; Timestamp lastMaxTime; public SynWork(ComboPooledDataSource dataSource, String sql, AtomicBoolean inWork) throws PropertyVetoException, SQLException { this.cpds = new ComboPooledDataSource(); this.cpds.setDriverClass(dataSource.getDriverClass()); cpds.setJdbcUrl(dataSource.getJdbcUrl()); cpds.setUser(dataSource.getUser()); cpds.setPassword(dataSource.getPassword()); cpds.setMinPoolSize(dataSource.getMinPoolSize()); cpds.setAcquireIncrement(dataSource.getAcquireIncrement()); cpds.setMaxPoolSize(dataSource.getMaxPoolSize()); cpds.setMaxStatements(dataSource.getMaxStatements()); this.sql = sql; this.inWork = inWork; lastMaxTime = new DBUtil().getMax(new QueryRunner(cpds), "PZ_RQ"); } @Override public void run() { try { //init(); if (inWork.get()) { return; } inWork.set(true); //获取当前日历 Calendar tmpCal = Calendar.getInstance(); //设置最早的同步时间 if (lastMaxTime == null) { lastMaxTime = new Timestamp((new SimpleDateFormat("yyyy-MM-dd hh:mm:ss").parse("1990-00-00 00:00:00")).getTime()); } //同步的起始时间 LocalDate startLocalDate = Instant.ofEpochMilli(lastMaxTime.getTime()).atZone(ZoneOffset.ofHours(8)).toLocalDate(); Timestamp now = new Timestamp(System.currentTimeMillis()); //同步的结束时间 LocalDate endLocalDate = Instant.ofEpochMilli(now.getTime()).atZone(ZoneOffset.ofHours(8)).toLocalDate(); //获取同步起始和结束的时间的间隔天数 long p = ChronoUnit.DAYS.between(startLocalDate, endLocalDate); //如果两个时间相差年数大于5天，就让上一次的时间加30天 if (p > 5) { tmpCal.setTimeInMillis(lastMaxTime.getTime()); tmpCal.add(Calendar.DAY_OF_YEAR, 30); now = new Timestamp(tmpCal.getTimeInMillis()); } //发出请求，获取数据 String json = getJson(lastMaxTime, now); System.out.println(json); //String json = FileUtils.readFileToString(new File("D:\\tmp\\response.json"), "utf-8"); if (json != null && !json.trim().isEmpty()) { //处理数据，解析数据，将解析完的数据入库 if (dealJson3(json)) { new DBUtil().delete(new QueryRunner(cpds), lastMaxTime); } } lastMaxTime = now; } catch (Exception e) { logger.error(e.getMessage()); e.printStackTrace(); } finally { //cpds.close(); inWork.set(false); } } public String getJson(Timestamp tstart, Timestamp tEnd) throws IOException { /* SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); String sStart = "begin=" + format.format(tstart); String sEnd = "end=" + format.format(tEnd); List<String> params = new ArrayList<>(); params.add(sStart); params.add(sEnd);*/ /*where=0%3D0 time=%3E1479427200000*/ SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); List<String> params = new ArrayList<>(); String where="&where=PZ_RQ<timestamp '"+format.format(tEnd)+"'"; //特殊符号转换 where=where.replaceAll(":", "%3A") .replaceAll(" ", "+") .replaceAll("'","%27") .replaceAll("<","%3C"); String time="&time=time>"+tstart.getTime(); time=time.replaceAll(">","%3E"); params.add(where); params.add(time); System.out.println(params); String url = "http://17.1.31.12/arcgis/rest/services/TDGY/TA_TDGY_GCS2000/MapServer/0/query?text=&objectIds=&geometry=&geometryType=esriGeometryEnvelope&inSR=&spatialRel=esriSpatialRelIntersects&relationParam=&outFields=*&returnGeometry=true&returnTrueCurves=false&maxAllowableOffset=&geometryPrecision=&outSR=&returnIdsOnly=false&returnCountOnly=false&orderByFields=&groupByFieldsForStatistics=&outStatistics=&returnZ=false&returnM=false&gdbVersion=&returnDistinctValues=false&resultOffset=&resultRecordCount=&queryByDistance=&returnExtentsOnly=false&datumTransformation=&parameterValues=&rangeValues=&f=pjson"; url+=where; url+=time; return RestUtil.httpGet2(url, null); } /** * 修改过后的---------可用 * * @param json * @return */ public boolean dealJson3(String json) throws ExecutionException, InterruptedException, ParseException, UnsupportedEncodingException { //获取请求的数据，只获取请求到的要如何库的数据 JSONObject jsonObject = JSON.parseObject(json); List<JSONObject> list = (List<JSONObject>) jsonObject.get("features"); List<Map<String, Object>> data = new Gson().fromJson(list.toString(), new TypeToken<List<Map<String, Object>>>() { }.getType()); if (data == null) { return false; } int submitnum = 0; CompletionService completionService = new ExecutorCompletionService<SynFinishResult>( SynThreadPool.getExec()); //批处理数据数量 int batchLen = CfgUtil.batchnum > data.size() ? data.size() : CfgUtil.batchnum; Object[][] nTmpParams = new Object[batchLen][]; int tmpct = 0; //遍历数据几点数据数组 for (int i = 0; i < data.size(); i++) { //具体的数据节点数据 Map<String, Object> map = (Map<String, Object>) data.get(i).get("attributes"); KvpMap nMap = new KvpMap(map); //要插入的sql参数 Object[] nTmp = new Object[CfgUtil.fields.size()]; //获取要插入的参数 for (int j = 0; j < CfgUtil.fields.size() - 1; j++) { /* clob字段，读取地理位置信息字段 if (CfgUtil.fields.get(j).getFldName().equalsIgnoreCase("aaa")) { nTmp[j] = new String(Base64.getDecoder().decode(((String)nMap.get(CfgUtil.fields.get(j).getFldName())).getBytes()), "utf-8"); } else */ //时间字段 if (CfgUtil.fields.get(j).getFldName().contains("PZ_RQ")) { if (nMap.get(CfgUtil.fields.get(j).getFldName()) != null) { nTmp[j] = new Tim