# 批處理中心 ## 基礎理論   [參考鏈接](https://gitee.com/kailing/partitionjob) 本工程不依賴mq,改為本地作業   # 設定讀取處理寫入規則 ``` @Bean("slaveStep") public Step slaveStep(DeliverPostProcessorItem processorItem, JdbcCursorItemReader reader) { CompositeItemProcessor itemProcessor = new CompositeItemProcessor(); List<ItemProcessor> processorList = new ArrayList<>(); processorList.add(processorItem); itemProcessor.setDelegates(processorList); return stepBuilderFactory.get("slaveStep") .<DeliverPost, DeliverPost>chunk(1000)//事務提交批次 .reader(reader) .processor(itemProcessor) .writer(dbItemWriter) .build(); } ``` ## 數據分片 ``` /** * @create 2019年4月2日 * Content :根據數據ID分片 */ public class ColumnRangePartitioner implements Partitioner { private JdbcOperations jdbcTemplate; ColumnRangePartitioner(DataSource dataSource){ this.jdbcTemplate = new JdbcTemplate(dataSource); } @Override public Map<String, ExecutionContext> partition(int gridSize) { Map<String, ExecutionContext> result = new LinkedHashMap<String, ExecutionContext>(); int current_thread = 1 ; int total_thread = gridSize ; while (current_thread <= total_thread) { ExecutionContext value = new ExecutionContext(); result.put("partition" + current_thread, value); value.putInt("current_thread", current_thread); value.putInt("total_thread", total_thread); current_thread++; } return result; } } ``` ## 本地基于游標方式讀取分片信息 ``` @Bean(destroyMethod = "") @StepScope public JdbcCursorItemReader<DeliverPost> JdbcCursorItemReader( @Value("#{stepExecutionContext['current_thread']}") Long current_thread, @Value("#{stepExecutionContext['total_thread']}") Long total_thread) { System.err.println("接收到分片參數["+total_thread+"->"+current_thread+"]"); JdbcCursorItemReader<DeliverPost> reader = new JdbcCursorItemReader<>(); reader.setDataSource(this.dataSource); // 設置數據源 reader.setFetchSize(100); // 設置一次最大讀取條數 reader.setRowMapper(new DeliverPostRowMapper()); // 把數據庫中的每條數據映射到Person對中 reader.setSql("select order_id , post_id from oc_deliver_post_t where post_id is not null and post_id <> '0' and mod(substring(order_id , -4) ,? )= ( ? -1 )"); reader.setPreparedStatementSetter(new PreparedStatementSetter() { public void setValues(PreparedStatement preparedStatement) throws SQLException { preparedStatement.setLong(1, total_thread); preparedStatement.setLong(2, current_thread); } }); return reader; } ``` ### 分片數據處理過程 ``` /** * @create 2019年4月2日 * Content :數據處理Item */ @Service public class DeliverPostProcessorItem implements ItemProcessor<DeliverPost, DeliverPost> { Logger logger = LoggerFactory.getLogger(DeliverPostProcessorItem.class); @Autowired private CommonDao commonDao ; @Autowired private ThirdServiceProp thirdServiceProp; @Override public DeliverPost process(DeliverPost deliverPost) throws Exception { logger.info("訂單號：【{}】經過處理器 ", deliverPost.getOrderId()); { // ems是否簽收 String resp = this.getEms(deliverPost.getPostId()); try { Map respMap = JSONObject.parseObject(resp, Map.class); if ("0000".equals(respMap.get("code"))) { Map rep = (Map) respMap.get("rep"); Map msg = (Map) rep.get("msg"); List<Map> traces = (List<Map>) msg.get("traces"); for (Iterator<Map> it = traces.iterator(); it.hasNext();) { Map temp = it.next(); if ("10".equals(temp.get("code"))) { // 已簽收 deliverPost.setIsArrived(1); } } } } catch (Exception e) { System.out.println(e); } } { // 中通是否簽收 String resp = this.getZT(deliverPost.getPostId()); try { Map respMap = JSONObject.parseObject(resp, Map.class); if ("0000".equals(respMap.get("code"))) { Map rep = (Map) respMap.get("rep"); Map msg = (Map) rep.get("msg"); List<Map> data = (List<Map>) msg.get("data"); for (Iterator<Map> it = data.iterator(); it.hasNext();) { Map temp = it.next(); List<Map> traces = (List<Map>) temp.get("traces"); for (Iterator<Map> it1 = traces.iterator(); it1.hasNext();) { Map tempT = it1.next(); if ("收件".equals(tempT.get("scanType"))) { // 已簽收 deliverPost.setIsArrived(1); } } } } } catch (Exception e) { } } return deliverPost; } public String getEms(String postId) { String transid= PointUtil.getRandom() ; // JSONObject resultJosn = JSONObject.fromObject(result); StringBuffer strbuf = new StringBuffer(); String jsonOut = ""; try { com.alibaba.fastjson.JSONObject obj = new com.alibaba.fastjson.JSONObject(); obj.put("method", "ems.inland.trace.query"); obj.put("action", "3th_ems"); Date date = new Date(); SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); obj.put("prea", sdf.format(date));// 163315236523523 com.alibaba.fastjson.JSONObject req = new com.alibaba.fastjson.JSONObject(); obj.put("req", req); com.alibaba.fastjson.JSONObject msg = new com.alibaba.fastjson.JSONObject(); req.put("msg", msg); msg.put("mailNo", postId); msg.put("authorization", "408a6c32e61d3ad5cb5c4e0cb3d2b089"); msg.put("timestamp", System.currentTimeMillis()); // 請求數據 jsonOut = obj.toString(); logger.info("EMS請求處理開始: transid=【{}】 ,req=【{}】", transid ,jsonOut); String callurl = commonDao.getHttpUrl("104"); int timeOut = 3000; URL url = new URL(thirdServiceProp.getUrl() + callurl); HttpURLConnection conn = (HttpURLConnection) url.openConnection(); conn.setRequestMethod("POST"); byte[] contentbyte = jsonOut.getBytes("UTF-8"); conn.setRequestProperty("Content-Type", "application/json;charset=UTF-8"); conn.setRequestProperty("Content-Length", contentbyte.length + ""); conn.setRequestProperty("Accept-Encoding", ""); conn.setRequestProperty("Accept", "application/json"); conn.setConnectTimeout(3000); conn.setReadTimeout(3000); conn.setUseCaches(false); conn.setDoInput(true); conn.setDoOutput(true); conn.connect(); OutputStream out = conn.getOutputStream(); out.write(contentbyte); // 發送請求報文 out.flush(); out.close(); InputStream in = conn.getInputStream(); BufferedReader dr = new BufferedReader(new InputStreamReader(in, "utf-8")); String text_rsp = null; while ((text_rsp = dr.readLine()) != null) { strbuf.append(text_rsp); } in.close(); logger.info("EMS請求處理結束: transid=【{}】 ,res=【{}】 ",transid, strbuf); } catch (Exception e) { strbuf.setLength(0); strbuf.append("{\"code\":\"8888\",\"detail\":\"失敗\"}"); logger.error(postId + "EMS轉發接口報錯！！！"); } return strbuf.toString(); } public String getZT(String postId) { String transid= PointUtil.getRandom() ; // JSONObject resultJosn = JSONObject.fromObject(result); StringBuffer strbuf = new StringBuffer(); String jsonOut = ""; try { com.alibaba.fastjson.JSONObject obj = new com.alibaba.fastjson.JSONObject(); obj.put("method", "api.traceInterfaceNewTraces"); obj.put("action", "3th_zto"); Date date = new Date(); SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); obj.put("prea", sdf.format(date));// 163315236523523 com.alibaba.fastjson.JSONObject req = new com.alibaba.fastjson.JSONObject(); obj.put("req", req); com.alibaba.fastjson.JSONObject msg = new com.alibaba.fastjson.JSONObject(); req.put("msg", msg); msg.put("company_id", "20f74746141c4433a15e7ddd5aade604"); msg.put("data", Arrays.asList(postId)); msg.put("msg_type", "NEW_TRACES"); // 請求數據 jsonOut = obj.toString(); logger.info("中通請求處理開始: transid=【{}】 ,req=【{}】 ",transid , jsonOut); String callurl = commonDao.getHttpUrl("103"); //固定token callurl =callurl.replace("tokenid", "798d3ed2ebaec83ae608c10207f783d6") ; int timeOut = 3000; URL url = new URL(thirdServiceProp.getUrl() + callurl); HttpURLConnection conn = (HttpURLConnection) url.openConnection(); conn.setRequestMethod("POST"); byte[] contentbyte = jsonOut.getBytes("UTF-8"); conn.setRequestProperty("Content-Type", "application/json;charset=UTF-8"); conn.setRequestProperty("Content-Length", contentbyte.length + ""); conn.setRequestProperty("Accept-Encoding", ""); conn.setRequestProperty("Accept", "application/json"); conn.setConnectTimeout(3000); conn.setReadTimeout(3000); conn.setUseCaches(false); conn.setDoInput(true); conn.setDoOutput(true); conn.connect(); OutputStream out = conn.getOutputStream(); out.write(contentbyte); // 發送請求報文 out.flush(); out.close(); InputStream in = conn.getInputStream(); BufferedReader dr = new BufferedReader(new InputStreamReader(in, "utf-8")); String text_rsp = null; while ((text_rsp = dr.readLine()) != null) { strbuf.append(text_rsp); } in.close(); logger.info("中通請求處理結束: transid=【{}】 ,res=【{}】 ",transid, strbuf); } catch (Exception e) { strbuf.setLength(0); strbuf.append("{\"code\":\"8888\",\"detail\":\"失敗\"}"); logger.error(postId + "中通轉發接口報錯！！！"); } return strbuf.toString(); } } ``` 分片數據輸出item ``` /** * @create 2019年4月2日 * Content :數據輸出item */ @Component @StepScope public class DBWriterItem<T> implements ItemWriter<T> { @Autowired private DeliverPostDao deliverPostDao ; @Override public void write(List<? extends T> list) throws Exception { deliverPostDao.batchInsert((List<? extends DeliverPost>) list); } } ```