异步新增修改记录

1.Controller：

@GetMapping("/abc")
    public Result getABC() {
        User u = getUser();
        Station station = getStation(u);
        String sip = station.getLvIp();
        String stationName = station.getStationName();
        String stationCode = station.getStationTelecode();
        String methodName = "/fangfaming";

//        JSONArray jsonArray = requestBureauService(methodName, sip);
        JSONArray jsonArray = requestService(methodName, stationName);

        List<Info> list = new ArrayList<>();
        Map<String, Info> comeMap = new HashMap<>();
        Date date = new Date();
        for (int i = 0; i < jsonArray.size(); i++) {
            Info info = new Info();
            JSONObject object = jsonArray.getJSONObject(i);
            UUID uuid = UUID.randomUUID();
            broadCast.setId(uuid.toString().replace("-", ""));
            info.setAreaId(object.getString("area_id"));
            info.setAreaNo(object.getString("area_no"));
            info.setAreaName(object.getString("area_name"));
            info.setAreaVolume(object.getString("area_volume"));
            info.setDefaultVolume(object.getString("default_volume"));
            info.setForeSigSrcId(object.getString("fore_sig_src_id"));
            info.setRelaTreeId(object.getString("rela_tree_id"));
            info.setParentId(object.getString("parent_id"));
            info.setAreaState(object.getString("area_state"));
            info.setStationName(object.getString("station_name"));
            info.setStationCode(stationCode);
            info.setCreateTime(date);
            info.setUpdateTime(date);
            list.add(info);
            comeMap.put(broadCast.getAreaId(), info);
        }
        if (list.size() > 0) {
            //DB数据
              List<Info> dbAll = broadCastInfoDao.findAllByStationName(stationName);

            //更新或新增
            InfoService.updateOrSave(comeMap, dbAll, stationName);
        }
        return Result.ok();
    }

    private JSONArray requestService(String methodName, String stationName) {
        JSONArray result = null;
        String response = "";
        try {
            String lvIp = getStation().getLvIp();

            Map<String, Object> param = new HashMap<>();
            param.put("methodName", "getPaBasInfo");
            param.put("companyName", "ks");
            param.put("userName", "kerry");
            param.put("passWord", "123456");
            param.put("userType", "Request");
//            param.put("stationName", stationName);
            param.put("stationName", "stationName");

            if (enableBureauService) {
                response = restUtil.httpPost(url + methodName, JSON.toJSONString(param));
            } else {
                response = restUtil.httpPost(lvIp + methodName, JSON.toJSONString(param));
            }

            //解析响应的数据
            JSONObject jsonObject = JSONObject.parseObject(response);
//            JSONArray data = jsonObject.getJSONArray("data");

            //如启动中转服务则多包装一层data
            if (enableBureauService) {
                JSONObject data1 = jsonObject.getJSONObject("data");
                result = data1.getJSONArray("data");
            } else {
                result = jsonObject.getJSONArray("data");
            }

//            //解析响应的数据
//            JSONObject jsonObject = JSONObject.parseObject(response);
//            JSONObject data = jsonObject.getJSONObject("data");
//
//            //如启动中转服务则多包装一层data
//            if (enableBureauService) {
//                JSONObject data1 = data.getJSONObject("data");
//                result = data1.getJSONArray("data");
//            } else {
//                result = data.getJSONArray("data");
//            }

        } catch (Exception e) {
            e.printStackTrace();
        }
        return result;
    }
}

public String httpPost(String url, String sendMsg) throws Exception {
    SimpleClientHttpRequestFactory requestFactory = new SimpleClientHttpRequestFactory();
    // 设置全局rest超时时长
    requestFactory.setConnectTimeout(retryTime);
    requestFactory.setReadTimeout(retryTime);
    RestTemplate restTemplate = new RestTemplate(requestFactory);
    // 设置请求头
    restTemplate.getMessageConverters()
            .add(0, new StringHttpMessageConverter(StandardCharsets.UTF_8));
    restTemplate.setErrorHandler(new DefaultResponseErrorHandler());
    HttpHeaders requestHeaders = new HttpHeaders();
    requestHeaders.setContentType(MediaType.APPLICATION_JSON);
    HttpEntity<String> requestEntity = new HttpEntity<>(sendMsg, requestHeaders);
    log.info("调用接口地址为：【{}】,入参为：【{}】", url, sendMsg);
    String httpPost = null;
    try {
        httpPost = restTemplate.postForObject(url, requestEntity, String.class);
    } catch (RestClientException e) {
        log.error("调用接口:{} 失败",url);
    }
    log.info("调用接口原始响应数据为：【{}】", httpPost);
    return httpPost;
}

2.Service：

void updateOrSave(Map<String, Info> comeMap, List<Info> dbAll, String stationName);

3.ServiceImpl：

@Service
public class InfoServiceImpl implements InfoService {

    @Autowired
    private ABCDao InfoDao;

    @Override
    public void updateOrSave(Map<String, Info> comeMap, List<Info> dbAll, String stationName) {

        Set<String> dbAreaIds = dbAll.stream().map(Info::getAreaId).collect(Collectors.toSet());

        CompletableFuture<Void> updateFuture = CompletableFuture.runAsync(() -> {
            //1. 更新操作
            //查出获取的需要更新到数据库的数据
            List<Info> updateList = comeMap.keySet().stream()
                    .filter(dbAreaIds::contains)
                    .map(comeMap::get).collect(Collectors.toList());

            updateList.forEach(broadCastInfo -> {
                //数据库中的待更新对象
                Info dbBroadCast = InfoDao.findByAreaIdAndStationName(Info.getAreaId(), Info.getStationName());

                //使用获取的更新对象的非空值去覆盖被更新的对象
                StringUtils.copyPropertiesIgnoreNull(broadCastInfo, dbBroadCast, "id", "createTime");

                //更新
                InfoDao.save(dbBroadCast);
            });
        });

        CompletableFuture<Void> saveFuture = CompletableFuture.runAsync(() -> {
            //2. 新增操作
            //查出获取的需要新增到数据库的数据
            List<Info> saveList = comeMap.keySet().stream()
                    .filter(areaId -> !dbAreaIds.contains(areaId))
                    .map(comeMap::get).collect(Collectors.toList());

            //保存
            InfoDao.saveAll(saveList);
        });

        //等待异步执行完成
        CompletableFuture.allOf(updateFuture, saveFuture).join();
    }
}

4.Dao:

public interface InfoDao extends BaseDao<BroadCastInfo, String> {

    List<Info> findAllByStationName(String stationName);
    Info findByAreaIdAndStationName(String areaId, String stationName);
}

5.Entity:

@Data
@Entity
@Table(name = "broad_cast_info")
public class Info implements Serializable {

    private static final long serialVersionUID = 1L;

    /**
     * 主键id
     */
    @Id
    @Column(length = 50,name = "s_id")
    private String id;

    /**
     * id
     */
    @Column(name = "area_id")
    private String areaId;

    @Column(name = "area_no")
    private String areaNo;

    @Column(name = "area_name")
    private String areaName;

    @Column(name = "area_volume")
    private String areaVolume;

    @Column(name = "default_volume")
    private String defaultVolume;

    @Column(name = "fore_sig_src_id")
    private String foreSigSrcId;

    @Column(name = "rela_tree_id")
    private String relaTreeId;

    @Column(name = "parent_id")
    private String parentId;

    @Column(name = "area_state")
    private String areaState;

    @Column(name = "station_name")
    private String stationName;

    @Column(name = "station_code")
    private String stationCode;

    @Column(name = "create_time")
    private Date createTime;

    @Column(name = "update_time")
    private Date updateTime;
}