SpringBoot定時任務實現數據同步的方法

Posted on by

本文實例為大傢分享瞭SpringBoot定時任務實現數據同步的具體代碼，供大傢參考，具體內容如下

前言

業務的需求是，通過中臺調用api接口獲得，設備數據，要求現實設備數據的同步。

方案一：通過輪詢接口的方式執行 pullData（） 方法實現數據同步

該方式的原理是先清空之前的所有數據，然後重新插入通過api調用獲取的最新數據。該方法的優點，邏輯簡單。缺點是，頻繁刪除、插入數據。再調用查詢數據時候，某一時刻，數據全部刪除，還沒及時插入的時候。數據可能有異常。

方案二：通過輪詢接口的方式執行 pullDataNew（） 方法實現數據同步

該方式的原理是先查詢數據庫，已有數據，然後和通過api調用獲取的最新數據進行比對，找出數據中增量、減量和變量，進行同步更新。該方法的優點，減少對數據庫的頻繁操作，提升性能。缺點：無發現明顯缺點。

package com.hxtx.spacedata.task;
 
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import com.google.api.client.util.Lists;
import com.hxtx.spacedata.common.domain.ResponseDTO;
import com.hxtx.spacedata.config.SpringContextUtil;
import com.hxtx.spacedata.controller.file.FilesMinioController;
import com.hxtx.spacedata.domain.entity.entityconfig.EntityPointEntity;
import com.hxtx.spacedata.service.entityconfig.EntityPointService;
import com.hxtx.spacedata.util.HttpProxyUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
 
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.stream.Collectors;
 
 
/**
 * 中臺設備數據 定時任務執行
 *
 * @author Tarzan Liu
 * @version 1.0.0
 * @description
 * @date 2020/12/07
 */
@Component
@Slf4j
public class EntityPointTask {
 
    @Autowired
    private EntityPointService entityPointService;
 
    @Value("${middleGround.server.host}")
    private String host;
 
    @Value("${middleGround.server.port}")
    private String port;
 
    private static FilesMinioController filesMinioController = SpringContextUtil.getBean(FilesMinioController.class);
 
    /**
     * 設備定義點數據拉取
     *
     * @author tarzan Liu
     * @date 2020/12/2
     */
    @Scheduled(cron = "0/30 * * * * ?") // 30秒校驗一次
    public void pullDataTaskByCorn() {
        String result = HttpProxyUtil.sendGet("http://" + host + ":" + port + "/interface/system/list");
        JSONObject jsonObject = JSON.parseObject(result);
        if (Objects.nonNull(jsonObject)) {
            JSONArray array = jsonObject.getJSONArray("data");
            if (array != null && array.size() != 0) {
                for (int i = 0; i < array.size(); i++) {
                    JSONObject obj = array.getJSONObject(i);
                    String systemId = obj.getString("id");
                    pullDataNew(systemId);
                }
            }
        }
    }
 
 
    @Transactional(rollbackFor = Throwable.class)
    public ResponseDTO<String> pullData(String code) {
        List<EntityPointEntity> list = Lists.newArrayList();
        String result = HttpProxyUtil.sendGet("http://" + host + ":" + port + "/interface/defintionView/listBySystemId/" + code);
        JSONObject jsonObject = JSON.parseObject(result);
        if (Objects.nonNull(jsonObject)) {
            JSONArray array = jsonObject.getJSONArray("data");
            if (array != null && array.size() != 0) {
                for (int i = 0; i < array.size(); i++) {
                    JSONObject obj = array.getJSONObject(i);
                    String pointId = obj.getString("pointId");
                    String name = obj.getString("name");
                    list.add(EntityPointEntity.builder().pointId(pointId).name(name).code(code).build());
                }
                List<EntityPointEntity> existList = entityPointService.list(new LambdaQueryWrapper<EntityPointEntity>().eq(EntityPointEntity::getCode, code).isNotNull(EntityPointEntity::getValue));
                if (CollectionUtils.isNotEmpty(existList)) {
                    Map<String, String> existMap = existList.stream().collect(Collectors.toMap(EntityPointEntity::getPointId, EntityPointEntity::getValue));
                    list.forEach(e -> {
                        String value = existMap.get(e.getPointId());
                        if (value != null) {
                            e.setValue(value);
                        }
                    });
                }
                entityPointService.remove(new LambdaQueryWrapper<EntityPointEntity>().eq(EntityPointEntity::getCode, code));
                entityPointService.saveBatch(list);
            }
        }
        return ResponseDTO.succ();
    }
 
 
    @Transactional(rollbackFor = Throwable.class)
    public ResponseDTO<String> pullDataNew(String code) {
        String result = HttpProxyUtil.sendGet("http://" + host + ":" + port + "/interface/defintionView/listBySystemId/" + code);
        JSONObject jsonObject = JSON.parseObject(result);
        if (Objects.nonNull(jsonObject)) {
            JSONArray data = jsonObject.getJSONArray("data");
            List<EntityPointEntity> list = data.toJavaList(EntityPointEntity.class);
            if (CollectionUtils.isNotEmpty(list)) {
                list.forEach(e -> e.setCode(code));
                List<EntityPointEntity> existList = entityPointService.list(new LambdaQueryWrapper<EntityPointEntity>().eq(EntityPointEntity::getCode, code));
                if (CollectionUtils.isNotEmpty(existList)) {
                    //存在map
                    Map<String, String> existMap = existList.stream().collect(Collectors.toMap(EntityPointEntity::getPointId, EntityPointEntity::getName));
                    //傳輸map
                    Map<String, String> dataMap = list.stream().collect(Collectors.toMap(EntityPointEntity::getPointId, EntityPointEntity::getName));
                    //增量
                    List<EntityPointEntity> increment = list.stream().filter(e -> existMap.get(e.getPointId()) == null).collect(Collectors.toList());
                    if (CollectionUtils.isNotEmpty(increment)) {
                        entityPointService.saveBatch(increment);
                    }
                    //減量
                    List<EntityPointEntity> decrement = existList.stream().filter(e -> dataMap.get(e.getPointId()) == null).collect(Collectors.toList());
                    if (CollectionUtils.isNotEmpty(decrement)) {
                        entityPointService.removeByIds(decrement.stream().map(EntityPointEntity::getId).collect(Collectors.toList()));
                    }
                    //變量
                    List<EntityPointEntity> variable = existList.stream().filter(e -> dataMap.get(e.getPointId()) != null && !dataMap.get(e.getPointId()).equals(e.getName())).collect(Collectors.toList());
                    if (CollectionUtils.isNotEmpty(variable)) {
                        variable.forEach(e -> {
                            e.setName(dataMap.get(e.getPointId()));
                        });
                        entityPointService.updateBatchById(variable);
                    }
                } else {
                    entityPointService.saveBatch(list);
                }
            }
        }
        return ResponseDTO.succ();
    }
 
 
}

數據庫對應實體類 

import com.baomidou.mybatisplus.annotation.IdType;
import com.baomidou.mybatisplus.annotation.TableId;
import com.baomidou.mybatisplus.annotation.TableName;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
 
import java.io.Serializable;
import java.util.Date;
 
@Builder
@NoArgsConstructor
@AllArgsConstructor
@Data
@TableName(value = "t_entity_point")
public class EntityPointEntity implements Serializable {
 
    private static final long serialVersionUID = 2181036545424452651L;
 
    /**
     * 定義點id
     */
    @TableId(value = "id", type = IdType.ASSIGN_ID)
    private Long id;
 
    /**
     * 定義點id
     */
    private String pointId;
 
    /**
     * 名稱
     */
    private String name;
 
    /**
     * 繪制數據
     */
    private String value;
 
    /**
     * 編碼
     */
    private String code;
 
    /**
     * 創建時間
     */
    private Date createTime;
 
}

HTTP請求代理工具類 

import lombok.extern.slf4j.Slf4j;
import org.apache.http.Consts;
import org.apache.http.HttpEntity;
import org.apache.http.HttpStatus;
import org.apache.http.NameValuePair;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.entity.UrlEncodedFormEntity;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.conn.ssl.SSLConnectionSocketFactory;
import org.apache.http.conn.ssl.TrustStrategy;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.message.BasicNameValuePair;
import org.apache.http.ssl.SSLContextBuilder;
import org.apache.http.util.EntityUtils;
 
import javax.net.ssl.SSLContext;
import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.URL;
import java.net.URLConnection;
import java.security.cert.CertificateException;
import java.security.cert.X509Certificate;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
 
 
/**
 * HTTP請求代理類
 *
 * @author tarzan Liu
 * @description 發送Get Post請求
 */
@Slf4j
public class HttpProxyUtil {
 
    /**
     * 使用URLConnection進行GET請求
     *
     * @param api_url
     * @return
     */
    public static String sendGet(String api_url) {
        return sendGet(api_url, "", "utf-8");
    }
 
    /**
     * 使用URLConnection進行GET請求
     *
     * @param api_url
     * @param param
     * @return
     */
    public static String sendGet(String api_url, String param) {
        return sendGet(api_url, param, "utf-8");
    }
 
    /**
     * 使用URLConnection進行GET請求
     *
     * @param api_url 請求路徑
     * @param param   請求格式有name1=value1&name2=value2、json、xml、map或其他形式，具體要看接收方的取值, 可以為空
     * @param charset 字符集
     * @return
     */
    public static String sendGet(String api_url, String param, String charset) {
        StringBuffer buffer = new StringBuffer();
        try {
            // 判斷有無參數，若是拼接好的url,就不必再拼接瞭
            if (param != null && !"".equals(param)) {
                api_url = api_url + "?" + param;
            }
            log.info("請求的路徑是：" + api_url);
            URL realUrl = new URL(api_url);
            // 打開聯接
            URLConnection conn = realUrl.openConnection();
            // 設置通用的請求屬性
            conn.setRequestProperty("accept", "*/*");
            conn.setRequestProperty("connection", "Keep-Alive");
            conn.setRequestProperty("user-agent", "Mozilla/4.0(compatible; MSIE 6.0; Windows NT 5.1; SV1)");
            conn.setConnectTimeout(12000);    //設置連接主機超時（單位：毫秒）
            conn.setReadTimeout(12000);    // 設置從主機讀取數據超時（單位：毫秒）
            conn.connect();    // 建立實際的聯接
 
            // 定義 BufferedReader輸入流來讀取URL的相應
            try (BufferedReader in = new BufferedReader(new InputStreamReader(conn.getInputStream(), charset))) {
                String line;
                while ((line = in.readLine()) != null) {
//                    buffer.append("\n"+line);
                    buffer.append(line);
                }
            }
        } catch (Exception e) {
            log.error("發送GET請求出現異常! " + e.getMessage());
            return null;
        }
        //  log.info("響應返回數據：" + buffer.toString());
        return buffer.toString();
    }
 
 
    /**
     * 使用URLConnection進行POST請求
     *
     * @param api_url 請求路徑
     * @param param   請求格式有name1=value1&name2=value2、json、xml、map或其他形式，具體要看接收方的取值,最好不為空
     * @return
     */
    public static String sendPost(String api_url, String param) {
        return sendPost(api_url, param, "utf-8");
    }
 
    /**
     * 使用URLConnection進行POST請求
     *
     * @param api_url 請求路徑
     * @param param   請求格式有name1=value1&name2=value2、json、xml、map或其他形式，具體要看接收方的取值,最好不為空
     * @param charset 字符集
     * @return
     */
    public static String sendPost(String api_url, String param, String charset) {
        StringBuffer buffer = new StringBuffer();
        try {
            log.info("請求的路徑是：" + api_url + ",參數是：" + param);
 
            URL realUrl = new URL(api_url);
            // 打開聯接
            URLConnection conn = realUrl.openConnection();
            // 設置通用的請求屬性
            conn.setRequestProperty("accept", "*/*");
            conn.setRequestProperty("connection", "Keep-Alive");
            conn.setRequestProperty("user-agent", "Mozilla/4.0(compatible; MSIE 6.0; Windows NT 5.1; SV1)");
            conn.setConnectTimeout(12000);    //設置連接主機超時（單位：毫秒）
            conn.setReadTimeout(12000);    // 設置從主機讀取數據超時（單位：毫秒）
 
            // 發送POST請求必須設置如下兩行
            conn.setDoOutput(true);
            conn.setDoInput(true);
 
            // 獲取URLConnection對象對應的輸出流
            try (PrintWriter out = new PrintWriter(conn.getOutputStream())) {
                out.print(param); // 發送請求參數
                out.flush();// flush輸出流的緩沖
            }
            // 定義 BufferedReader輸入流來讀取URL的相應,得指明使用UTF-8編碼，否則到API服務器XML的中文不能被成功識別
            try (BufferedReader in = new BufferedReader(new InputStreamReader(conn.getInputStream(), charset))) {
                String line;
                while ((line = in.readLine()) != null) {
//                    buffer.append("\n"+line);
                    buffer.append(line);
                }
            }
        } catch (Exception e) {
            log.error("發送POST請求出現異常! " + e.getMessage());
            e.printStackTrace();
        }
        log.info("響應返回數據：" + buffer.toString());
        return buffer.toString();
    }
 
    public static CloseableHttpClient createSSLClientDefault() throws Exception {
        SSLContext sslContext = new SSLContextBuilder().loadTrustMaterial(null, new AllTrustStrategy()).build();
        SSLConnectionSocketFactory sslSf = new SSLConnectionSocketFactory(sslContext);
        return HttpClients.custom().setSSLSocketFactory(sslSf).build();
    }
 
    // 加載證書
    private static class AllTrustStrategy implements TrustStrategy {
        public boolean isTrusted(X509Certificate[] x509Certificates, String s) throws CertificateException {
            return true;
        }
    }
 
    /**
     * 支持https請求
     *
     * @param url
     * @param param
     * @return
     * @throws Exception
     */
    public static String sendHttpClientPost(String url, Map<String, String> param) throws Exception {
        CloseableHttpClient httpClient = createSSLClientDefault();
        HttpPost httpPost = null;
        CloseableHttpResponse response = null;
        String result = "";
        try {
            // 發起HTTP的POST請求
            httpPost = new HttpPost(url);
            List<NameValuePair> paramList = new ArrayList<NameValuePair>();
            for (String key : param.keySet()) {
                paramList.add(new BasicNameValuePair(key, param.get(key)));
            }
            log.info("http請求地址：" + url + ",參數：" + paramList.toString());
            // UTF8+URL編碼
            httpPost.setEntity(new UrlEncodedFormEntity(paramList, Consts.UTF_8));
            httpPost.setConfig(RequestConfig.custom().setConnectTimeout(30000).setSocketTimeout(30000).build());
            response = httpClient.execute(httpPost);
            HttpEntity entity = response.getEntity();
            int statusCode = response.getStatusLine().getStatusCode();
            if (HttpStatus.SC_OK == statusCode) { // 如果響應碼是200
 
            }
            result = EntityUtils.toString(entity);
            log.info("狀態碼：" + statusCode + ",響應信息：" + result);
        } finally {
            if (response != null) {
                response.close();
            }
            if (httpPost != null) {
                httpPost.releaseConnection();
            }
            httpClient.close();
        }
        return result;
    }
}

以上就是本文的全部內容，希望對大傢的學習有所幫助，也希望大傢多多支持WalkonNet。

推薦閱讀：