[Java知识库]Spring Boot实现数据同步

问题描述：

实际项目开发过程中，需要将第三方数据库中的部门同步到当前部门下，两个数据库部门表的字段有很多不同。

方法一

Controller层：

@PostMapping("/syncDepartmentMsg")
    public ResultNoEncryptVo<Object> syncDepartmentMsg() {

        Integer line = departmentService.syncDepartmentMsg();
        return ResultNoEncryptVo.ok(line);
    }

Service层：

   Integer syncDepartmentMsg();

ServiceImpl层：

@Override
    public Integer syncDepartmentMsg() {

        HttpResultDepartmentVo httpResultDepartmentVo;
        int isSuccess = 1;

        try {
            String departmentInfo = HttpClient.httpGet(syncDepartmentInfo,"29");
            httpResultDepartmentVo = JSONObject.parseObject(departmentInfo, HttpResultDepartmentVo.class);
            List<DepartmentRemoteVo> data = httpResultDepartmentVo.getData();
            departmentMapper.insertOrUpdateDepartmentBatch(data);
        } catch (Exception e) {
            e.printStackTrace();
            isSuccess = 0;
            log.error("同步用户数据异常: orgId -> {}, catch ->",  e);
        }

        return isSuccess;
    }

Mapper层

@Mapper
public interface DepartmentMapper extends BaseMapper<DepartmentVo> {

    void insertOrUpdateDepartmentBatch(@Param("departmentRemoteVo") List<DepartmentRemoteVo> data);
}

xml:

<insert id="insertOrUpdateDepartmentBatch">
        insert into `answer_department`( id,....
        )
        values
        <foreach collection="departmentRemoteVo" item="departmentVo" separator=",">
            (
            #{departmentVo.id},
           ......
            )
        </foreach>
        ON DUPLICATE KEY UPDATE sort = values(sort), bind_user_id = values(bind_user_id), cate_id = values(cate_id),
       .....

方法二

@RequestMapping("/dataSynchronization")
public class DataSynchronizationController extends BaseManageController{
    /**
     * 部门同步接口
     */
//    @Scheduled(cron = "0 */3 * * * ?") // 三分钟执行一次
//    @Scheduled(cron = "0/10 * * * * ?") // 三分钟执行一次  定时任务
    @Transactional(rollbackFor = Throwable.class)
    public void pullDataNew() {
        String result = HttpClient.httpGet("http://app.cq.qiludev.com/v1/osps/app/department/getGovaskDepartment?fieldType=all","29");
        JSONObject jsonObject = JSON.parseObject(result);
        log.info(jsonObject.toJSONString());
        if (Objects.nonNull(jsonObject)) {
            JSONArray data = jsonObject.getJSONArray("data");
            List<DepartmentRemoteVo> list = data.toJavaList(DepartmentRemoteVo.class);
            if (CollectionUtils.isNotEmpty(list)) {

                //查询数据库已有的数据
                List<DepartmentVo> existList = departmentService.list();
                if (CollectionUtils.isNotEmpty(existList)) {
                    //存在map
                    Map<Integer, DepartmentVo> existMap = existList.stream().collect(Collectors.toMap(DepartmentVo::getId,DepartmentVo->DepartmentVo));
                    //传输map
                    Map<Integer, DepartmentRemoteVo> dataMap = list.stream().collect(Collectors.toMap(DepartmentRemoteVo::getId, DepartmentRemoteVo->DepartmentRemoteVo));

                    List<DepartmentVo> variable = existList.stream().filter(department -> dataMap.get(department.getId()) != null
                            && !dataMap.get(department.getId()).getDepartment().equals(department.getDepartment())
                    ).collect(Collectors.toList());
                    if (CollectionUtils.isNotEmpty(variable)) {
                        variable.forEach(department -> {
                            departmentService.removeById(department.getId());
                            log.info("删除的id");
                        });
                    }
                    //增量
                    List<DepartmentRemoteVo> increment = list.stream().filter(department -> existMap.get(department.getId()) == null).collect(Collectors.toList());
                    if (CollectionUtils.isNotEmpty(increment)) {
                        departmentRemoteService.saveBatch(increment);
                    }
                } else {
                    departmentRemoteService.saveBatch(list);
                }
            }
        }

    }

工具类：
HTTPClient连接池

@Component
public class HttpClient {

    /**
     * 池化管理
     */
    private static PoolingHttpClientConnectionManager poolConnManager = null;
    private static CloseableHttpClient httpClient;
    /**
     * 请求器配置
     */
    private static RequestConfig requestConfig;

    static {
        SSLContextBuilder builder = new SSLContextBuilder();
        try {
            builder.loadTrustMaterial(null, new TrustSelfSignedStrategy());
            SSLConnectionSocketFactory sslsf = new SSLConnectionSocketFactory(builder.build());

            // 配置同时支持HTTP和HTTPS协议
            Registry<ConnectionSocketFactory> socketFactoryRegistry = RegistryBuilder.<ConnectionSocketFactory>create()
                    .register("http", PlainConnectionSocketFactory.getSocketFactory()).register("https", sslsf).build();
            // 初始化连接管理器
            poolConnManager = new PoolingHttpClientConnectionManager(socketFactoryRegistry);
            // 设置最大连接数
            poolConnManager.setMaxTotal(200);
            // 设置路由
            poolConnManager.setDefaultMaxPerRoute(50);
            // 设置默认超时限制初始化requestConfig
            int socketTimeout = 3000;
            int connectTimeout = 3000;
            int connectionRequestTimeout = 3000;
            requestConfig = RequestConfig.custom().setConnectionRequestTimeout(connectionRequestTimeout)
                    .setSocketTimeout(socketTimeout).setConnectTimeout(connectTimeout).build();
            // 初始化HttpClient
            httpClient = getConnection();
            log.info("HttpClient init ..");
        } catch (NoSuchAlgorithmException | KeyStoreException | KeyManagementException e) {
            e.printStackTrace();
        }

    }

    /**
     * 创建HttpClient对象
     *
     */
    private static CloseableHttpClient getConnection() {
        return HttpClients.custom().setConnectionManager(poolConnManager)
                // 设置请求配置
                .setDefaultRequestConfig(requestConfig)
                // 设置重试次数
                .setRetryHandler(new DefaultHttpRequestRetryHandler(0, false))
                .build();
    }

    //  get  //

    /**
     * HttpGet请求方法 - 基本
     *
     */
    public static String httpGet(String url) {

        HttpGet httpGet = new HttpGet(url);
        CloseableHttpResponse response = null;
        String result = "";
        try {
            response = httpClient.execute(httpGet);
            HttpEntity entity = response.getEntity();
            result = EntityUtils.toString(entity, "utf-8");
            EntityUtils.consume(entity);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (response != null) {
                try {
                    response.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
        return result;
    }

    /**
     * HttpGet请求方法 - 携带Token
     *
     */
    public static String httpGet(String url, String orgid) {

        HttpGet httpGet = new HttpGet(url);
        httpGet.setHeader("orgId", orgid);
        CloseableHttpResponse response = null;
        String result = "";
        try {
            response = httpClient.execute(httpGet);
            HttpEntity entity = response.getEntity();
            result = EntityUtils.toString(entity, "utf-8");
            EntityUtils.consume(entity);
            log.info("请求(httpGet-token) -> {}", url);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (response != null) {
                try {
                    response.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
        return result;
    }

    /**
     * HttpGet请求方法 - 请求一县一端
     *
     */
    public static String doGet(String url) {

        HttpGet httpGet = new HttpGet(url);
        httpGet.setHeader("Dubbo-Attachments", "appkey=comment,appsecret=d2ff9b3583507c7e48ed3b1769b89f7d");
        httpGet.setHeader("Accept", "application/json,*/*");
        CloseableHttpResponse response = null;
        String result = "";
        long start = System.currentTimeMillis();
        try {
            response = httpClient.execute(httpGet);
            HttpEntity entity = response.getEntity();
            result = EntityUtils.toString(entity, "utf-8");
            EntityUtils.consume(entity);
            log.info("请求(httpGet) -> {}", url);
        } catch (IOException e) {
            long end = System.currentTimeMillis() - start;
            log.error("请求报错(httpGet) -> {}, 耗时 -> {}ms", url, end);
            e.printStackTrace();
        } finally {
            if (response != null) {
                try {
                    response.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
        return result;
    }

    /**
     * HttpGet请求方法 
     *
     */
    public static String doGet(String url,String orgid,String token) {

        HttpGet httpGet = new HttpGet(url);
        httpGet.setHeader("Dubbo-Attachments", "appkey=comment,appsecret=d2ff9b3583507c7e48ed3b1769b89f7d");
        /// 后期用这个, 目前测试赞用评论
///        httpGet.setHeader("Dubbo-Attachments", "appkey=listen,appsecret=d36e4ac13a1e8b7eacab7a952578a938");
        httpGet.setHeader("Accept", "application/json,*/*");
        httpGet.setHeader("orgid", orgid);
        httpGet.setHeader("CQ-TOKEN", token);
        CloseableHttpResponse response = null;
        String result = "";
        try {
            response = httpClient.execute(httpGet);
            HttpEntity entity = response.getEntity();
            result = EntityUtils.toString(entity, "utf-8");
            EntityUtils.consume(entity);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (response != null) {
                try {
                    response.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
        return result;
    }

    //  POST  //

    /**
     * HttpPost请求
     *
     */
    public static String doPost(String url, Map<String, Object> paramsMap) {

        CloseableHttpResponse closeableHttpResponse = null;
        //配置连接超时时间
        RequestConfig requestConfig = RequestConfig.custom()
                .setConnectTimeout(3000)
                .setConnectionRequestTimeout(3000)
                .setSocketTimeout(3000)
                .setRedirectsEnabled(true)
                .build();
        HttpPost httpPost = new HttpPost(url);
        //设置超时时间
        httpPost.setConfig(requestConfig);

        //装配post请求参数
        List<NameValuePair> list = new ArrayList<>();
        for (String key : paramsMap.keySet()) {
            list.add(new BasicNameValuePair(key, String.valueOf(paramsMap.get(key))));
        }

        try {
            //将参数进行编码为合适的格式,如将键值对编码为param1=value1&param2=value2
            UrlEncodedFormEntity urlEncodedFormEntity = new UrlEncodedFormEntity(list, "utf-8");
            httpPost.setEntity(urlEncodedFormEntity);

            //执行 post请求
            closeableHttpResponse = httpClient.execute(httpPost);
            String strRequest = "";
            if (null != closeableHttpResponse) {
///                System.out.println(closeableHttpResponse.getStatusLine().getStatusCode());
                if (closeableHttpResponse.getStatusLine().getStatusCode() == HttpStatus.SC_OK) {
                    HttpEntity httpEntity = closeableHttpResponse.getEntity();
                    strRequest = EntityUtils.toString(httpEntity);
                } else {
                    strRequest = "Error Response" + closeableHttpResponse.getStatusLine().getStatusCode();
                }
            }
            return strRequest;

        } catch (ClientProtocolException e) {
            e.printStackTrace();
            return "协议异常";
        } catch (ParseException e) {
            e.printStackTrace();
            return "解析异常";
        } catch (IOException e) {
            e.printStackTrace();
            return "传输异常";
        } finally {
            if (closeableHttpResponse != null) {
                try {
                    closeableHttpResponse.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }

        }
    }

}