读取CSV文件

@Data
@Builder//构造器
public class CtripCommentReviewId {
    private String reviewId;
}

public class CtripCommentCleanByFileTask {

    public static final String file_path = "/home/q/commondata/ctrip/";

    public static final String WRAPPER_ID = "CtripImport";

    @Autowired
    private CrawlerReviewDetailDao crawlerReviewDetailDao;

    @Autowired
    private CommentImportService commentImportService;

    @Autowired
    private ReviewMemcacheServiceImpl reviewMemcacheService;
    private static final Function<String[], Optional<CtripCommentReviewId>> CSV = (input) -> Optional.of(CtripCommentReviewId.builder().reviewId(input[0]).build());

    @QSchedule("hotel.ugc.review_task.ctrip.comment.clean.by.file")
    public void execute(Parameter parameter) {
        long startTime = System.currentTimeMillis();
        String fileName = parameter.getString("file");
        if (StringUtils.isBlank(fileName)) {
            log.warn("parameters is blank");
            return;
        }
        File file = new File(String.format("%s%s", file_path, fileName));
        try {
            queryCommentsByFile(file, Charsets.UTF_8, CSV);
        } catch (IOException e) {
            log.warn("exception:{}", e);
            MonitorUtils.monitor(MonitorUtils.CTRIP_COMMENT_FILE_EXCEPTION);
            return;
        }
        MonitorUtils.success(MonitorUtils.CTRIP_COMMENT_CLEAN_BY_FILE, startTime);
    }

    private void queryCommentsByFile(File file, Charset charset, Function<String[], Optional<CtripCommentReviewId>> lineParseFunction) throws IOException {
        try (CSVReader csvReader = new CSVReader(new InputStreamReader(new FileInputStream(file), charset))) {
            for (String[] line = csvReader.readNext(); line != null; line = csvReader.readNext()) {
                //Funtion()----->>>将一个字符串数组（第一个元素）转化成对象
                Optional<CtripCommentReviewId> commentOptional = lineParseFunction.apply(line);
                if (commentOptional != null && commentOptional.isPresent()) {
                    CtripCommentReviewId ctripCommentReviewId = commentOptional.get();
                    if(ctripCommentReviewId==null){
                        continue;
                    }
                    dealComment(ctripCommentReviewId.getReviewId());
                }
            }
        } catch (IOException e) {
            log.warn(e.getMessage(), e);
            throw e;
        }
    }

　　

读取json数据：CtripCommentExt commentExt = Json2Utils.readValue(detail.getParameters(), CtripCommentExt.class);

写入json数据：函数changeStatus(detail.getId(), Json2Utils.writeValueAsString(commentExt));-----》函数中内容：crawlerReviewDetailDao.updateParameters(id, parameters)

 public void dealComment(String reviewId) {
        if (StringUtils.isEmpty(reviewId)) {
            return;
        }
        CrawlerReviewDetail detail = null;
        String seq = "";
        try {
            detail = crawlerReviewDetailDao.queryByReviewIdAndRapperId(reviewId, WRAPPER_ID);
            if (detail == null) {
                return;
            }
            CtripCommentExt commentExt = Json2Utils.readValue(detail.getParameters(), CtripCommentExt.class);
            if (commentExt == null || commentExt.getExt() == null || commentExt.getComment() == null) {
                log.warn("invalid content id={}", detail.getId());
                MonitorUtils.monitor(MonitorUtils.CTRIP_COMMENT_INVALID_CONTENT_BY_CLEAN);
                return;
            }
            seq = detail.getHotelSeq();
            if (StringUtils.isEmpty(seq)) {
                return;
            }
            ReviewSynchronization synchronization = commentExt.getExt().getSynchronization();
            /**
             * status=1 2  3  cid !=0 删除comment 改状态 || cid=0 改状态
             * status=0  cid!=0 删除comment cid=0不做任何操作
             */
            if (synchronization == null) {
                return;
            }
            long cid = synchronization.getQunarCid();
            int status = synchronization.getStatus();
            if (ReviewSyncStatus.NEED_UPDATE.getCode() == status || ReviewSyncStatus.NEED_INSERT.getCode() == status ||
                    ReviewSyncStatus.NEED_DELETE.getCode() == status) {
                if (cid != 0) {
                    deleteComment(seq, cid);
                }
                synchronization.setStatus(ReviewSyncStatus.SYNC_DONE.getCode());
                changeStatus(detail.getId(), Json2Utils.writeValueAsString(commentExt));
            } else {
                if (cid != 0){
                    deleteComment(seq, cid);
                }
            }
            MonitorUtils.success(MonitorUtils.CTRIP_COMMENT_CLEAN);
        } catch (Exception e) {
            log.warn("reviewId:{},exception:{}", reviewId, e);
        }
    }