ÁÄÁÄPowerJobµÄGridFsManager

@Slf4j
@Service
public class GridFsManager implements InitializingBean {

    private final Environment environment;

    private final MongoDatabase db;

    private boolean available;

    private final Map<String, GridFSBucket> bucketCache = Maps.newConcurrentMap();

    public static final String LOG_BUCKET = "log";

    public static final String CONTAINER_BUCKET = "container";

    public GridFsManager(Environment environment, @Autowired(required = false) MongoTemplate mongoTemplate) {
        this.environment = environment;
        if (mongoTemplate != null) {
            this.db = mongoTemplate.getDb();
        } else {
            this.db = null;
        }
    }

    /**
     * ÊÇ·ñ¿ÉÓÃ
     * @return true£º¿ÉÓã»false£º²»¿ÉÓÃ
     */
    public boolean available() {
        return available;
    }

    //......

    private GridFSBucket getBucket(String bucketName) {
        return bucketCache.computeIfAbsent(bucketName, ignore -> GridFSBuckets.create(db, bucketName));
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        String enable = environment.getProperty(PowerJobServerConfigKey.MONGODB_ENABLE, Boolean.FALSE.toString());
        available = Boolean.TRUE.toString().equals(enable) && db != null;
        log.info("[GridFsManager] available: {}, db: {}", available, db);
    }
}    

    /**
     * ´æ´¢Îļþµ½ GridFS
     * @param localFile ±¾µØÎļþ
     * @param bucketName Í°Ãû³Æ
     * @param fileName GirdFSÖеÄÎļþÃû³Æ
     * @throws IOException Òì³£
     */
    public void store(File localFile, String bucketName, String fileName) throws IOException {
        if (available()) {
            GridFSBucket bucket = getBucket(bucketName);
            try (BufferedInputStream bis = new BufferedInputStream(new FileInputStream(localFile))) {
                bucket.uploadFromStream(fileName, bis);
            }
        }
    }

    /**
     * ´Ó GridFS ÏÂÔØÎļþ
     * @param targetFile ÏÂÔصÄÄ¿±êÎļþ£¨±¾µØÎļþ£©
     * @param bucketName Í°Ãû³Æ
     * @param fileName GirdFSÖеÄÎļþÃû³Æ
     * @throws IOException Òì³£
     */
    public void download(File targetFile, String bucketName, String fileName) throws IOException {
        if (available()) {
            GridFSBucket bucket = getBucket(bucketName);
            try (GridFSDownloadStream gis = bucket.openDownloadStream(fileName);
                 BufferedOutputStream bos = new BufferedOutputStream(new FileOutputStream(targetFile))
            ) {
                byte[] buffer = new byte[1024];
                int bytes = 0;
                while ((bytes = gis.read(buffer)) != -1) {
                    bos.write(buffer, 0, bytes);
                }
                bos.flush();
            }
        }
    }

    /**
     * ɾ³ý¼¸ÌìÇ°µÄÎļþ
     * @param bucketName Í°Ãû³Æ
     * @param day ÈÕÆÚÆ«ÒÆÁ¿£¬µ¥Î» Ìì
     */
    public void deleteBefore(String bucketName, int day) {

        Stopwatch sw = Stopwatch.createStarted();

        Date date = DateUtils.addDays(new Date(), -day);
        GridFSBucket bucket = getBucket(bucketName);
        Bson filter = Filters.lt("uploadDate", date);

        // Ñ­»·É¾³ýÐÔÄܺܲÎÒ²ÂÄã¿Ï¶¨Ã»¿´¹ý¹Ù·½ÊµÏÖ[¹·Í·]£ºorg.springframework.data.mongodb.gridfs.GridFsTemplate.delete
        bucket.find(filter).forEach((Consumer<GridFSFile>) gridFSFile -> {
            ObjectId objectId = gridFSFile.getObjectId();
            try {
                bucket.delete(objectId);
                log.info("[GridFsManager] deleted {}#{}", bucketName, objectId);
            }catch (Exception e) {
                log.error("[GridFsManager] deleted {}#{} failed.", bucketName, objectId, e);
            }
        });
        log.info("[GridFsManager] clean bucket({}) successfully, delete all files before {}, using {}.", bucketName, date, sw.stop());
    }

    public boolean exists(String bucketName, String fileName) {
        GridFSBucket bucket = getBucket(bucketName);
        GridFSFindIterable files = bucket.find(Filters.eq("filename", fileName));
        try {
            GridFSFile first = files.first();
            return first != null;
        }catch (Exception ignore) {
        }
        return false;
    }

    @Async(PJThreadPool.BACKGROUND_POOL)
    public void sync(Long instanceId) {

        Stopwatch sw = Stopwatch.createStarted();
        try {
            // Ïȳ־û¯µ½±¾µØÎļþ
            File stableLogFile = genStableLogFile(instanceId);
            // ½«ÎļþÍÆË͵½ MongoDB
            if (gridFsManager.available()) {
                try {
                    gridFsManager.store(stableLogFile, GridFsManager.LOG_BUCKET, genMongoFileName(instanceId));
                    log.info("[InstanceLog-{}] push local instanceLogs to mongoDB succeed, using: {}.", instanceId, sw.stop());
                }catch (Exception e) {
                    log.warn("[InstanceLog-{}] push local instanceLogs to mongoDB failed.", instanceId, e);
                }
            }
        }catch (Exception e) {
            log.warn("[InstanceLog-{}] sync local instanceLogs failed.", instanceId, e);
        }
        // ɾ³ý±¾µØÊý¾Ý¿âÊý¾Ý
        try {
            instanceId2LastReportTime.remove(instanceId);
            CommonUtils.executeWithRetry0(() -> localInstanceLogRepository.deleteByInstanceId(instanceId));
            log.info("[InstanceLog-{}] delete local instanceLog successfully.", instanceId);
        }catch (Exception e) {
            log.warn("[InstanceLog-{}] delete local instanceLog failed.", instanceId, e);
        }
    }