webcrawler/web/src/main/java/web/service/DataService.java

package web.service;

import core.service.BaseService;
import db.model.CidEntity;
import db.model.DataModel;
import org.apache.commons.io.FileUtils;
import org.apache.commons.lang3.StringUtils;
import org.hibernate.Session;
import org.hibernate.query.NativeQuery;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Future;

@Service
public class DataService extends BaseService {

    private ThreadPoolTaskExecutor executor;

    private boolean isStop;

    public void start(Integer threadSize){
        isStop=false;
        executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(threadSize);
        executor.setMaxPoolSize(threadSize);
        executor.setQueueCapacity(0);
        executor.setWaitForTasksToCompleteOnShutdown(true);
        executor.initialize();
        log.info("开始");
        new Thread(()->{
            getCid(threadSize);
        }).start();
    }

    public void stop(){
        if(executor!=null) {
            executor.shutdown();
            isStop = true;
        }
    }

    private void getCid(Integer threadSize) {


        File file = new File("result");
        int i = 0;
        try {
            String s = FileUtils.readFileToString(file, "UTF-8");
            if (StringUtils.isNumeric(s)) {
                i = Integer.parseInt(s);
            }
        } catch (IOException e) {
            log.error(e);
        }
        for (; i <= 15717791; ) {

            if(isStop){
                log.info("停止");
                break;
            }

            List<Future<Boolean>> tasks = new ArrayList<>();
            LinkedList<Integer> cid = new LinkedList<>();
            ConcurrentLinkedQueue<Integer> success = new ConcurrentLinkedQueue<>();
            for (int j = 0; j < threadSize; j++) {
                cid.add(i++);
            }
            do {
                Integer c=cid.poll();
                if(c!=null) {
                    try {
                        executor.execute(() -> {
                            save(c);
                            success.add(c);
                        });
                    } catch (Exception e) {
                        cid.add(c);
                    }
                }
            } while (success.size() < threadSize);
            tasks.clear();
            cid.clear();
            success.clear();
            try {
                FileUtils.writeStringToFile(file, i + "", "UTF-8", false);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }


    private void save(Integer id) {
        try {
            DataModel dataModel = baseDao1.get(DataModel.class, id);
            CidEntity cidEntity = baseDao1.get(CidEntity.class, id);

            if (dataModel != null && cidEntity != null && cidEntity.getTypeid() != null) {
                dataModel.setTypeId(cidEntity.getTypeid());
                baseDao2.executeWithNativeSession((Session session) -> {
                    session.beginTransaction();
                    int i = 1;
                    String sql = "insert into data_" + cidEntity.getTypeid() + " (aid, author, backup_vid, cache, cover, dispatch, dispatch_servers, dp_done, dp_done_flv, dp_done_hdmp4, dp_done_mp4, duration, files, letv_addr, letv_vid, letv_vu, mid, page, storage, storage_server, subtitle, title, type, type_id, upload, upload_meta, vid, vp, cid) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
                    NativeQuery query = session.createNativeQuery(sql)
                            .setParameter(i++, dataModel.getAid()).setParameter(i++, dataModel.getAuthor()).setParameter(i++, dataModel.getBackupVid()).setParameter(i++, dataModel.getCache()).setParameter(i++, dataModel.getCover())
                            .setParameter(i++, dataModel.getDispatch()).setParameter(i++, dataModel.getDispatchServers()).setParameter(i++, dataModel.getDpDone()).setParameter(i++, dataModel.getDpDoneFlv()).setParameter(i++, dataModel.getDpDoneHdmp4()).setParameter(i++, dataModel.getDpDoneMp4())
                            .setParameter(i++, dataModel.getDuration()).setParameter(i++, dataModel.getFiles()).setParameter(i++, dataModel.getLetvAddr()).setParameter(i++, dataModel.getLetvVid()).setParameter(i++, dataModel.getLetvVu()).setParameter(i++, dataModel.getMid())
                            .setParameter(i++, dataModel.getPage()).setParameter(i++, dataModel.getStorage()).setParameter(i++, dataModel.getStorageServer()).setParameter(i++, dataModel.getSubtitle()).setParameter(i++, dataModel.getTitle()).setParameter(i++, dataModel.getType())
                            .setParameter(i++, dataModel.getTypeId()).setParameter(i++, dataModel.getUpload()).setParameter(i++, dataModel.getUploadMeta()).setParameter(i++, dataModel.getVid()).setParameter(i++, dataModel.getVp()).setParameter(i++, dataModel.getCid());
                    return query.executeUpdate();
                });

            }
        }catch (Exception e){
            log.error(e);
        }

    }
}