XCKJ/irc-netcore-api
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

增加内存优先级队列、后端上传的地方都设置类别，和业务参数、增加启动后台恢复队列任务

Browse Source
This commit is contained in:
hang
2026-03-11 16:46:00 +08:00
parent 59be897fd8
commit 3bca306702
28 changed files with 2630 additions and 232 deletions
Download Patch File Download Diff File Expand all files Collapse all files
IRaCIS.Core.Application/Helper/OSSService.cs
+180 -42
View File
@@ -7,6 +7,8 @@ using Amazon.S3;
using Amazon.S3.Model;
using Amazon.SecurityToken;
using Amazon.SecurityToken.Model;
using IRaCIS.Core.Application.Interfaces;
using IRaCIS.Core.Application.ViewModel;
using IRaCIS.Core.Infrastructure;
using IRaCIS.Core.Infrastructure.NewtonsoftJson;
using MassTransit;
@@ -16,7 +18,9 @@ using Minio;
using Minio.DataModel;
using Minio.DataModel.Args;
using Minio.Exceptions;
using Org.BouncyCastle.Tls;
using Serilog.Parsing;
using SkiaSharp;
using System.IO;
using System.Reactive.Linq;
using System.Runtime.InteropServices;
@@ -83,6 +87,8 @@ public class ObjectStoreServiceOptions
public bool IsOpenStoreSync { get; set; }
public string ApiDeployRegion { get; set; }
public List<SyncStoreConfig> SyncConfigList { get; set; } = new List<SyncStoreConfig>();
}
@@ -91,10 +97,16 @@ public class SyncStoreConfig
{
public string Domain { get; set; }
public string UploadRegion { get; set; }
public string TargetRegion { get; set; }
public string Primary { get; set; }
public string Target { get; set; }
public bool IsOpenSync { get; set; }
}
public class ObjectStoreDTO
@@ -156,6 +168,7 @@ public enum ObjectStoreUse
AWS = 2,
}
#endregion
// aws 参考链接 https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/dotnetv3/S3/S3_Basics
@@ -166,8 +179,8 @@ public interface IOSSService
public Task RestoreFilesByPrefixAsync(string prefix, int restoreDays = 3, int batchSize = 100);
public Task<string> UploadToOSSAsync(Stream fileStream, string oosFolderPath, string fileRealName, bool isFileNameAddGuid = true);
public Task<string> UploadToOSSAsync(string localFilePath, string oosFolderPath, bool isFileNameAddGuid = true, bool randomFileName = false);
public Task<string> UploadToOSSAsync(Stream fileStream, string oosFolderPath, string fileRealName, bool isFileNameAddGuid = true, FileUploadRecordAddOrEdit? uploadInfo = null);
public Task<string> UploadToOSSAsync(string localFilePath, string oosFolderPath, bool isFileNameAddGuid = true, bool randomFileName = false, FileUploadRecordAddOrEdit? uploadInfo = null);
public Task DownLoadFromOSSAsync(string ossRelativePath, string localFilePath);
@@ -183,28 +196,27 @@ public interface IOSSService
List<string> GetRootFolderNames();
public ObjectStoreDTO GetObjectStoreTempToken(string? domain = null);
public ObjectStoreDTO GetObjectStoreTempToken(string? domain = null, bool? isGetAllTempToken = null);
public Task MoveObject(string sourcePath, string destPath, bool overwrite = true);
public Task<long> GetObjectSizeAsync(string sourcePath);
public Task SyncFileAsync(string objectKey, ObjectStoreUse source, ObjectStoreUse destination, CancellationToken ct = default);
}
public class OSSService : IOSSService
public class OSSService(IOptionsMonitor<ObjectStoreServiceOptions> options,
IFileUploadRecordService _fileUploadRecordService) : IOSSService
{
public ObjectStoreServiceOptions ObjectStoreServiceOptions { get; set; }
public ObjectStoreServiceOptions ObjectStoreServiceOptions { get; set; } = options.CurrentValue;
private AliyunOSSTempToken AliyunOSSTempToken { get; set; }
private AWSTempToken AWSTempToken { get; set; }
public OSSService(IOptionsMonitor<ObjectStoreServiceOptions> options)
{
ObjectStoreServiceOptions = options.CurrentValue;
}
/// <summary>
/// 将指定前缀下的所有现有文件立即转为目标存储类型
@@ -719,6 +731,9 @@ public class OSSService : IOSSService
}
}
/// <summary>
/// oosFolderPath 不要 "/ "开头 应该： TempFolder/ChildFolder
/// </summary>
@@ -726,8 +741,9 @@ public class OSSService : IOSSService
/// <param name="oosFolderPath"></param>
/// <param name="fileRealName"></param>
/// <param name="isFileNameAddGuid"></param>
/// <param name="uploadInfo"> 只用赋值业务参数Id 和批次信息即可，其他信息不用传递</param>
/// <returns></returns>
public async Task<string> UploadToOSSAsync(Stream fileStream, string oosFolderPath, string fileRealName, bool isFileNameAddGuid = true)
public async Task<string> UploadToOSSAsync(Stream fileStream, string oosFolderPath, string fileRealName, bool isFileNameAddGuid = true, FileUploadRecordAddOrEdit? uploadInfo = null)
{
BackBatchGetToken();
@@ -735,13 +751,8 @@ public class OSSService : IOSSService
try
{
using (var memoryStream = new MemoryStream())
{
fileStream.Seek(0, SeekOrigin.Begin);
fileStream.CopyTo(memoryStream);
memoryStream.Seek(0, SeekOrigin.Begin);
if (fileStream.CanSeek)
fileStream.Seek(0, SeekOrigin.Begin);
if (ObjectStoreServiceOptions.ObjectStoreUse == "AliyunOSS")
@@ -753,7 +764,7 @@ public class OSSService : IOSSService
// 上传文件
var result = _ossClient.PutObject(aliConfig.BucketName, ossRelativePath, memoryStream);
var result = _ossClient.PutObject(aliConfig.BucketName, ossRelativePath, fileStream);
}
else if (ObjectStoreServiceOptions.ObjectStoreUse == "MinIO")
@@ -768,8 +779,8 @@ public class OSSService : IOSSService
var putObjectArgs = new PutObjectArgs()
.WithBucket(minIOConfig.BucketName)
.WithObject(ossRelativePath)
.WithStreamData(memoryStream)
.WithObjectSize(memoryStream.Length);
.WithStreamData(fileStream)
.WithObjectSize(fileStream.Length);
await minioClient.PutObjectAsync(putObjectArgs);
}
@@ -793,7 +804,7 @@ public class OSSService : IOSSService
var putObjectRequest = new Amazon.S3.Model.PutObjectRequest()
{
BucketName = awsConfig.BucketName,
InputStream = memoryStream,
InputStream = fileStream,
Key = ossRelativePath,
};
@@ -803,7 +814,6 @@ public class OSSService : IOSSService
{
throw new BusinessValidationFailedException("未定义的存储介质类型");
}
}
}
catch (Exception ex)
{
@@ -812,13 +822,27 @@ public class OSSService : IOSSService
}
var returnPath = "/" + ossRelativePath;
return "/" + ossRelativePath;
if (ObjectStoreServiceOptions.IsOpenStoreSync && uploadInfo != null)
{
uploadInfo.FileSize = fileStream.CanSeek ? fileStream.Length : 0;
uploadInfo.Path = returnPath;
uploadInfo.FileName = fileRealName;
uploadInfo.FileType = Path.GetExtension(returnPath);
await _fileUploadRecordService.AddOrUpdateFileUploadRecord(uploadInfo);
}
return returnPath;
}
//后端批量上传 或者下载，不每个文件获取临时token
private void BackBatchGetToken()
{
@@ -860,10 +884,12 @@ public class OSSService : IOSSService
/// <param name="randomFileName">随机文件名</param>
/// <returns></returns>
/// <exception cref="BusinessValidationFailedException"></exception>
public async Task<string> UploadToOSSAsync(string localFilePath, string oosFolderPath, bool isFileNameAddGuid = true, bool randomFileName = false)
public async Task<string> UploadToOSSAsync(string localFilePath, string oosFolderPath, bool isFileNameAddGuid = true, bool randomFileName = false, FileUploadRecordAddOrEdit? uploadInfo = null)
{
BackBatchGetToken();
long fileSize = 0;
var localFileName = Path.GetFileName(localFilePath);
var ossRelativePath = isFileNameAddGuid ? $"{oosFolderPath}/{Guid.NewGuid()}_{localFileName}" : $"{oosFolderPath}/{localFileName}";
@@ -883,6 +909,8 @@ public class OSSService : IOSSService
// 上传文件
var result = _ossClient.PutObject(aliConfig.BucketName, ossRelativePath, localFilePath);
fileSize = result.ContentLength;
}
else if (ObjectStoreServiceOptions.ObjectStoreUse == "MinIO")
{
@@ -898,7 +926,9 @@ public class OSSService : IOSSService
.WithObject(ossRelativePath)
.WithFileName(localFilePath);
await minioClient.PutObjectAsync(putObjectArgs);
var result = await minioClient.PutObjectAsync(putObjectArgs);
fileSize = result.Size;
}
else if (ObjectStoreServiceOptions.ObjectStoreUse == "AWS")
{
@@ -923,14 +953,30 @@ public class OSSService : IOSSService
Key = ossRelativePath,
};
await amazonS3Client.PutObjectAsync(putObjectRequest);
var result = await amazonS3Client.PutObjectAsync(putObjectRequest);
fileSize = result.ContentLength;
}
else
{
throw new BusinessValidationFailedException("未定义的存储介质类型");
}
return "/" + ossRelativePath;
var returnPath = "/" + ossRelativePath;
if (ObjectStoreServiceOptions.IsOpenStoreSync && uploadInfo != null)
{
uploadInfo.FileSize = fileSize;
uploadInfo.Path = returnPath;
uploadInfo.FileName = Path.GetFileName(localFilePath);
uploadInfo.FileType = Path.GetExtension(returnPath);
await _fileUploadRecordService.AddOrUpdateFileUploadRecord(uploadInfo);
}
return returnPath;
}
@@ -1042,11 +1088,6 @@ public class OSSService : IOSSService
// 直接返回流
return result.Content;
//// 将OSS返回的流复制到内存流中并返回
//var memoryStream = new MemoryStream();
//await result.Content.CopyToAsync(memoryStream);
//memoryStream.Position = 0; // 重置位置以便读取
//return memoryStream;
}
else if (ObjectStoreServiceOptions.ObjectStoreUse == "MinIO")
{
@@ -1112,10 +1153,6 @@ public class OSSService : IOSSService
// ⭐ 直接返回流
return response.ResponseStream;
//var memoryStream = new MemoryStream();
//await response.ResponseStream.CopyToAsync(memoryStream);
//memoryStream.Position = 0;
//return memoryStream;
}
else
{
@@ -1459,8 +1496,31 @@ public class OSSService : IOSSService
/// <returns></returns>
public async Task DeleteFromPrefix(string prefix, bool isCache = false)
{
GetObjectStoreTempToken();
//打开了同步的，删除的时候，一起删除
if (ObjectStoreServiceOptions.IsOpenStoreSync && ObjectStoreServiceOptions.SyncConfigList.Any(t => t.IsOpenSync))
{
foreach (var config in ObjectStoreServiceOptions.SyncConfigList.Where(t => t.IsOpenSync))
{
ObjectStoreServiceOptions.ObjectStoreUse = config.Primary;
GetObjectStoreTempToken();
await DeleteFromPrefixInternal(prefix, isCache);
}
}
else
{
GetObjectStoreTempToken();
await DeleteFromPrefixInternal(prefix, isCache);
}
}
private async Task DeleteFromPrefixInternal(string prefix, bool isCache = false)
{
if (ObjectStoreServiceOptions.ObjectStoreUse == "AliyunOSS")
{
var aliConfig = ObjectStoreServiceOptions.AliyunOSS;
@@ -1610,8 +1670,32 @@ public class OSSService : IOSSService
}
}
public async Task DeleteObjects(List<string> objectKeys, bool isCache = false)
{
//打开了同步的，删除的时候，一起删除
if (ObjectStoreServiceOptions.IsOpenStoreSync && ObjectStoreServiceOptions.SyncConfigList.Any(t => t.IsOpenSync))
{
foreach (var config in ObjectStoreServiceOptions.SyncConfigList.Where(t => t.IsOpenSync))
{
ObjectStoreServiceOptions.ObjectStoreUse = config.Primary;
GetObjectStoreTempToken();
await DeleteObjectsInternal(objectKeys, isCache);
}
}
else
{
GetObjectStoreTempToken();
await DeleteObjectsInternal(objectKeys, isCache);
}
}
public async Task DeleteObjectsInternal(List<string> objectKeys, bool isCache = false)
{
GetObjectStoreTempToken();
if (ObjectStoreServiceOptions.ObjectStoreUse == "AliyunOSS")
@@ -1774,7 +1858,7 @@ public class OSSService : IOSSService
public ObjectStoreDTO GetObjectStoreTempToken(string? domain = null)
public ObjectStoreDTO GetObjectStoreTempToken(string? domain = null, bool? isGetAllTempToken = null)
{
//如果传递了域名，并且打开了存储同步，根据域名使用的具体存储覆盖之前的配置，否则就用固定的配置
if (ObjectStoreServiceOptions.IsOpenStoreSync && domain.IsNotNullOrEmpty())
@@ -1788,9 +1872,9 @@ public class OSSService : IOSSService
}
}
var objectStoreDTO = new ObjectStoreDTO() { ObjectStoreUse= ObjectStoreServiceOptions.ObjectStoreUse, IsOpenStoreSync = ObjectStoreServiceOptions.IsOpenStoreSync ,SyncConfigList= ObjectStoreServiceOptions .SyncConfigList};
var objectStoreDTO = new ObjectStoreDTO() { ObjectStoreUse = ObjectStoreServiceOptions.ObjectStoreUse, IsOpenStoreSync = ObjectStoreServiceOptions.IsOpenStoreSync, SyncConfigList = ObjectStoreServiceOptions.SyncConfigList };
if (ObjectStoreServiceOptions.ObjectStoreUse == "AliyunOSS")
if (ObjectStoreServiceOptions.ObjectStoreUse == "AliyunOSS" || isGetAllTempToken == true)
{
var ossOptions = ObjectStoreServiceOptions.AliyunOSS;
@@ -1842,7 +1926,7 @@ public class OSSService : IOSSService
{
objectStoreDTO.MinIO = ObjectStoreServiceOptions.MinIO;
}
else if (ObjectStoreServiceOptions.ObjectStoreUse == "AWS")
else if (ObjectStoreServiceOptions.ObjectStoreUse == "AWS" || isGetAllTempToken == true)
{
var awsOptions = ObjectStoreServiceOptions.AWS;
@@ -1895,4 +1979,58 @@ public class OSSService : IOSSService
return objectStoreDTO;
}
public async Task SyncFileAsync(string objectKey, ObjectStoreUse source, ObjectStoreUse destination, CancellationToken ct = default)
{
GetObjectStoreTempToken(isGetAllTempToken: true);
var aliConfig = ObjectStoreServiceOptions.AliyunOSS;
var _ossClient = new OssClient(RuntimeInformation.IsOSPlatform(OSPlatform.Windows) ? aliConfig.EndPoint : aliConfig.InternalEndpoint, AliyunOSSTempToken.AccessKeyId, AliyunOSSTempToken.AccessKeySecret, AliyunOSSTempToken.SecurityToken);
var awsConfig = ObjectStoreServiceOptions.AWS;
var credentials = new SessionAWSCredentials(AWSTempToken.AccessKeyId, AWSTempToken.SecretAccessKey, AWSTempToken.SessionToken);
//提供awsEndPoint（域名）进行访问配置
var clientConfig = new AmazonS3Config
{
RegionEndpoint = RegionEndpoint.GetBySystemName(awsConfig.Region)
};
var amazonS3Client = new AmazonS3Client(credentials, clientConfig);
// 根据源选择流式下载
Stream sourceStream = source switch
{
ObjectStoreUse.AliyunOSS => _ossClient.GetObject(aliConfig.BucketName, objectKey).Content,
ObjectStoreUse.AWS => (await amazonS3Client.GetObjectAsync(awsConfig.BucketName, objectKey, ct)).ResponseStream,
_ => throw new BusinessValidationFailedException("未定义的同步类型")
};
if (source == ObjectStoreUse.AliyunOSS)
{
var putRequest = new Amazon.S3.Model.PutObjectRequest
{
BucketName = awsConfig.BucketName,
Key = objectKey,
InputStream = sourceStream
};
await amazonS3Client.PutObjectAsync(putRequest, ct);
}
else if (source == ObjectStoreUse.AWS)
{
_ossClient.PutObject(aliConfig.BucketName, objectKey, sourceStream);
}
else
{
throw new BusinessValidationFailedException("未定义的同步类型");
}
await sourceStream.DisposeAsync(); // 释放流
}
}