PowerJob的GridFsManager工作流程源碼解讀
序
本文主要研究一下PowerJob的GridFsManager
GridFsManager
tech/powerjob/server/persistence/mongodb/GridFsManager.java
@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);
}
}GridFsManager實現(xiàn)了InitializingBean接口,其afterPropertiesSet從environment讀取oms.mongodb.enable配置,默認(rèn)為false;其構(gòu)造器注入mongoTemplate,若為null則available為false;其getBucket方法則根據(jù)bucketName緩存到bucketCache,若不存在則通過GridFSBuckets.create創(chuàng)建
store
/**
* 存儲文件到 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);
}
}
}store方法先獲取bucket,再讀取localFile,通過bucket.uploadFromStream上傳
download
/**
* 從 GridFS 下載文件
* @param targetFile 下載的目標(biāo)文件(本地文件)
* @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();
}
}
}download方法則先獲取bucket,再通過bucket.openDownloadStream獲取GridFSDownloadStream,最后寫到targetFile
deleteBefore
/**
* 刪除幾天前的文件
* @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);
// 循環(huán)刪除性能很差?我猜你肯定沒看過官方實現(xiàn)[狗頭]: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());
}deleteBefore主要通過bucket.find(Filters.lt("uploadDate", date))找到GridFSFile,再挨個執(zhí)行delete
exists
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;
}exists方法則通過bucket.find(Filters.eq("filename", fileName))來進(jìn)行查找
sync
tech/powerjob/server/core/instance/InstanceLogService.java
@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);
}
// 刪除本地數(shù)據(jù)庫數(shù)據(jù)
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);
}
}InstanceLogService的sync方法先持久化到本地文件,再將文件推送到 MongoDB
小結(jié)
GridFsManager實現(xiàn)了InitializingBean接口,其afterPropertiesSet從environment讀取oms.mongodb.enable配置,默認(rèn)為false;其構(gòu)造器注入mongoTemplate,若為null則available為false;其store和download方法都先判斷是否available,為false則空操作。
以上就是PowerJob的GridFsManager工作流程源碼解讀的詳細(xì)內(nèi)容,更多關(guān)于PowerJob GridFsManager的資料請關(guān)注腳本之家其它相關(guān)文章!
相關(guān)文章
java運(yùn)行時數(shù)據(jù)區(qū)域和類結(jié)構(gòu)詳解
這篇文章主要介紹了java運(yùn)行時數(shù)據(jù)區(qū)域和類結(jié)構(gòu),具有很好的參考價值,希望對大家有所幫助。如有錯誤或未考慮完全的地方,望不吝賜教2021-07-07
SpringBoot接受參數(shù)相關(guān)注解方式
SpringBoot接受參數(shù)的注解包括@PathVariable、@RequestHeader、@RequestParameter、@CookieValue、@RequestBody、@RequestAttribute和@SessionAttribute等,每個注解都有詳細(xì)的使用方法和示例代碼2024-12-12
SpringBoot父子線程數(shù)據(jù)傳遞的五種方案介紹
在實際開發(fā)過程中我們需要父子之間傳遞一些數(shù)據(jù),比如用戶信息等。該文章從5種解決方案解決父子之間數(shù)據(jù)傳遞困擾,文中通過示例代碼介紹的非常詳細(xì),對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值,需要的朋友們下面隨著小編來一起學(xué)習(xí)吧2022-09-09
Java service層獲取HttpServletRequest工具類的方法
今天小編就為大家分享一篇關(guān)于Java service層獲取HttpServletRequest工具類的方法,小編覺得內(nèi)容挺不錯的,現(xiàn)在分享給大家,具有很好的參考價值,需要的朋友一起跟隨小編來看看吧2018-12-12
Java中BeanUtils.copyProperties基本用法與小坑
本文主要介紹了Java中BeanUtils.copyProperties基本用法與小坑,文中通過示例代碼介紹的非常詳細(xì),對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2023-04-04
SpringBoot+Vue+Redis實現(xiàn)單點(diǎn)登錄(一處登錄另一處退出登錄)
小編接到一個需求,需要實現(xiàn)用戶在瀏覽器登錄后,跳轉(zhuǎn)到其他頁面,當(dāng)用戶在其它地方又登錄時,前面用戶登錄的頁面退出登錄,這篇文章主要介紹了SpringBoot+Vue+Redis實現(xiàn)單點(diǎn)登錄,需要的朋友可以參考下2019-12-12

