客戶的環境是一個網絡音樂播放系統,根據網絡狀況提供給手機用戶收聽各類碼率的MP3歌曲,在客戶沒購買歌曲的狀況下提供一個三十秒內的試聽版本。這樣一個系統很是明確地一個需求就是會按期須要將一批從音樂版商手中獲取到的高比特率音樂文件轉換成各類低碼率的MP3文件和試聽文件,因爲收到版商的文件數量和時間都不肯定,因此長期部署大量的轉碼服務器爲系統提供轉碼服務顯然很是浪費資源,可是若是不許備好足夠的轉碼服務器的話,當大批量文件須要轉碼時又無法可以快速完成任務,在如今這個時間比金錢更加劇要的互聯網時代顯然是不可接受的。這時候選擇公有云這樣高彈性、按需計費的計算平臺就顯得很是合適了。html
使用Azure Fuctions+Azure Batch+Azure Blob Storage方案,所有都是基於PaaS平臺,無需對服務器進行管理,省去服務器在平常維護中各類補丁安全管理要求。node
利用Azure Function監控Blob文件變化,Azure Functions的一大優勢就是提供了不一樣類型的觸發器(http Trigger,Blob Trigger,Timer Trigger,Queue Trigger…),這裏咱們正好利用上Blob Trigger用來監控Blob文件的變化。windows
首先是建立一個Azure Functions的Project安全
而後指定Function是用Blob Trigger的。服務器
建立ListeningBlob函數,網絡
using System; using System.Collections.Generic; using System.Diagnostics; using System.IO; using Microsoft.Azure.WebJobs; using Microsoft.Azure.WebJobs.Host; using Microsoft.Extensions.Logging; using Microsoft.WindowsAzure.Storage; using Microsoft.WindowsAzure.Storage.Blob; using Microsoft.WindowsAzure.Storage.Queue; namespace MS.CSU.mp3encoder { public static class ListeningBlob { static string key_Convert = Environment.GetEnvironmentVariable("KeyConvert") ?? "-i \"{0}\" -codec:a libmp3lame -b:a {1} \"{2}\" -y"; static string work_Dir = Path.GetTempPath(); static string targetStorageConnection = Environment.GetEnvironmentVariable("targetStorageConnection"); static string sourceStorageConnection = Environment.GetEnvironmentVariable("sourceStorageConnection"); static string bitRates = Environment.GetEnvironmentVariable("bitRates") ?? "192k;128k;64k"; static string keyPreview = Environment.GetEnvironmentVariable("keyPreview") ?? "-ss 0 -t 29 -i \"{0}\" \"{1}\""; static CloudBlobClient blobOutputClient; static string blobOutputContainerName = Environment.GetEnvironmentVariable("outputContainer") ?? "output"; static CloudBlobContainer blobOutputContainer; static CloudBlobClient blobInputClient; static CloudBlobContainer blobInputContainer; [FunctionName("ListeningBlob")] [return: Queue("jobs")] public static void Run([BlobTrigger("source/{name}", Connection = "sourceStorageConnection")]Stream myBlob, string name, Uri uri, TraceWriter log) { AzureBatch batch = new AzureBatch(sourceStorageConnection); //保證每一個音頻文件都有本身的處理文件夾,避免衝突 Guid jobId = Guid.NewGuid(); log.Info($"Job:{jobId},C# Blob trigger function Processed blob\n Name:{name} \n Size: {myBlob.Length} Bytes,Path:{uri.ToString()}"); //將源Blob剪切到TargetBlob,將源文件移出監控Blob容器,避免誤觸發 try { initBlobClient(); CloudBlockBlob sourceBlob = blobInputContainer.GetBlockBlobReference($"{name}"); name = Path.GetFileNameWithoutExtension(name); CloudBlockBlob targetBlob = blobOutputContainer.GetBlockBlobReference($"{name}_{jobId}/{name}.mp3"); targetBlob.StartCopy(sourceBlob); sourceBlob.Delete(); uri = targetBlob.Uri; } catch (Exception err) { log.Error($"刪除源Blob錯誤!Err:{err}"); return ; } List<EncodeJob> jobs = new List<EncodeJob>(); string url = Uri.EscapeUriString(uri.ToString()); log.Info($"須要轉換的碼率:{bitRates}"); string[] bitsRateNames = bitRates.Split(';'); Dictionary<string, bool> status = new Dictionary<string, bool>(); foreach (var s in bitsRateNames) { if (string.IsNullOrWhiteSpace(s)) continue; var job = new EncodeJob() { OutputName = $"{name}{s}.mp3", Name = name, Command = string.Format(key_Convert, name, s, $"{name}{s}.mp3"), id = jobId, InputUri = uri }; batch.QueueTask(job); } var previewJob = new EncodeJob() { Name = name, OutputName = $"{name}preview.mp3", Command = string.Format(keyPreview, name, $"{name}preview.mp3"), InputUri = uri, id = jobId, }; batch.QueueTask(previewJob); //Directory.Delete($"{work_Dir}\\{jobId}",true); } static void initBlobClient() { CloudStorageAccount storageOutputAccount = CloudStorageAccount.Parse(targetStorageConnection); // Create a blob client for interacting with the blob service. blobOutputClient = storageOutputAccount.CreateCloudBlobClient(); blobOutputContainer = blobOutputClient.GetContainerReference(blobOutputContainerName); blobOutputContainer.CreateIfNotExists(); //初始化輸入的Storage容器 CloudStorageAccount storageInputAccount = CloudStorageAccount.Parse(sourceStorageConnection); // Create a blob client for interacting with the blob service. blobInputClient = storageInputAccount.CreateCloudBlobClient(); blobInputContainer = blobInputClient.GetContainerReference("source"); } } }
建立Batch服務帳號,而且獲取Batch Account的相關信息。架構
到https://ffmpeg.zeranoe.com/下載最新的ffmpeg程序,安裝後將ffmpeg.exe單獨壓縮成zip文件,而後上傳到Batch中,爲程序調用作準備,app
構建Azure Batch類用於調用Azure Batch進行ffmpeg進行轉換async
using Microsoft.Azure.Batch; using Microsoft.Azure.Batch.Auth; using Microsoft.Azure.Batch.Common; using Microsoft.WindowsAzure.Storage; using Microsoft.WindowsAzure.Storage.Blob; using System; using System.Collections.Generic; using System.IO; using System.Linq; using System.Text; using System.Threading.Tasks; namespace MS.CSU.mp3encoder { public class AzureBatch { //ffmpeg相關信息; string env_appPackageInfo = Environment.GetEnvironmentVariable("ffmpegversion")??"ffmpeg 3.4"; string appPackageId = "ffmpeg"; string appPackageVersion = "3.4"; // Pool and Job constants private const string PoolId = "WinFFmpegPool"; private const int DedicatedNodeCount = 0; private const int LowPriorityNodeCount = 5; //指定執行轉碼任務的VM機型 private const string PoolVMSize = "Standard_F2"; private const string JobName = "WinFFmpegJob"; string outputStorageConnection; string outputContainerName = "output"; string batchAccount = Environment.GetEnvironmentVariable("batchAccount"); string batchKey = Environment.GetEnvironmentVariable("batchKey"); string batchAccountUrl = Environment.GetEnvironmentVariable("batchAccountUrl"); string strMaxTaskPerNode = Environment.GetEnvironmentVariable("MaxTaskPerNode") ?? "4"; //設置每一個計算節點能同時處理的任務數量,可根據選擇的VM類型和任務類型適當調整 int maxTaskPerNode=4; public AzureBatch(string storageConnection) { outputStorageConnection = storageConnection; } //用於單元測試時建立Batch對象 public AzureBatch(string storageConnection, string _batchAccount, string _batchAccountUrl, string _batchKey) { outputStorageConnection = storageConnection; batchAccount = _batchAccount; batchAccountUrl = _batchAccountUrl; batchKey = _batchKey; maxTaskPerNode = int.TryParse(strMaxTaskPerNode, out maxTaskPerNode) ? maxTaskPerNode : 4; appPackageId = env_appPackageInfo.Split(' ')[0] ?? "ffmpeg"; appPackageVersion = env_appPackageInfo.Split(' ')[1] ?? "3.4"; } /// <summary> /// Returns a shared access signature (SAS) URL providing the specified /// permissions to the specified container. The SAS URL provided is valid for 2 hours from /// the time this method is called. The container must already exist in Azure Storage. /// </summary> /// <param name="blobClient">A <see cref="CloudBlobClient"/>.</param> /// <param name="containerName">The name of the container for which a SAS URL will be obtained.</param> /// <param name="permissions">The permissions granted by the SAS URL.</param> /// <returns>A SAS URL providing the specified access to the container.</returns> private string GetContainerSasUrl(CloudBlobClient blobClient, string containerName, SharedAccessBlobPermissions permissions) { // Set the expiry time and permissions for the container access signature. In this case, no start time is specified, // so the shared access signature becomes valid immediately. Expiration is in 2 hours. SharedAccessBlobPolicy sasConstraints = new SharedAccessBlobPolicy { SharedAccessExpiryTime = DateTime.UtcNow.AddHours(2), Permissions = permissions }; // Generate the shared access signature on the container, setting the constraints directly on the signature CloudBlobContainer container = blobClient.GetContainerReference(containerName); string sasContainerToken = container.GetSharedAccessSignature(sasConstraints); // Return the URL string for the container, including the SAS token return String.Format("{0}{1}", container.Uri, sasContainerToken); } // BATCH CLIENT OPERATIONS - FUNCTION IMPLEMENTATIONS /// <summary> /// Creates the Batch pool. /// </summary> /// <param name="batchClient">A BatchClient object</param> /// <param name="poolId">ID of the CloudPool object to create.</param> private void CreatePoolIfNotExist(BatchClient batchClient, string poolId) { // if (batchClient.PoolOperations.GetPool(poolId) != null) // { // return; // } CloudPool pool = null; try { ImageReference imageReference = new ImageReference( publisher: "MicrosoftWindowsServer", offer: "WindowsServer", sku: "2012-R2-Datacenter-smalldisk", version: "latest"); //ImageReference imageReference = new ImageReference( // publisher: "MicrosoftWindowsServer", // offer: "WindowsServer", // sku: "2016-Datacenter-samlldisk", // version: "latest"); VirtualMachineConfiguration virtualMachineConfiguration = new VirtualMachineConfiguration( imageReference: imageReference, nodeAgentSkuId: "batch.node.windows amd64"); // Create an unbound pool. No pool is actually created in the Batch service until we call // CloudPool.Commit(). This CloudPool instance is therefore considered "unbound," and we can // modify its properties. pool = batchClient.PoolOperations.CreatePool( poolId: poolId, targetDedicatedComputeNodes: DedicatedNodeCount, targetLowPriorityComputeNodes: LowPriorityNodeCount, virtualMachineSize: PoolVMSize, virtualMachineConfiguration: virtualMachineConfiguration); pool.MaxTasksPerComputeNode = maxTaskPerNode; // Specify the application and version to install on the compute nodes // This assumes that a Windows 64-bit zipfile of ffmpeg has been added to Batch account // with Application Id of "ffmpeg" and Version of "3.4". // Download the zipfile https://ffmpeg.zeranoe.com/builds/win64/static/ffmpeg-3.4-win64-static.zip // to upload as application package pool.ApplicationPackageReferences = new List<ApplicationPackageReference> { new ApplicationPackageReference { ApplicationId = appPackageId, Version = appPackageVersion } }; pool.Commit(); } catch (BatchException be) { // Accept the specific error code PoolExists as that is expected if the pool already exists if (be.RequestInformation?.BatchError?.Code == BatchErrorCodeStrings.PoolExists) { // Console.WriteLine("The pool {0} already existed when we tried to create it", poolId); } else { throw; // Any other exception is unexpected } } } /// <summary> /// Creates a job in the specified pool. /// </summary> /// <param name="batchClient">A BatchClient object.</param> /// <param name="jobId">ID of the job to create.</param> /// <param name="poolId">ID of the CloudPool object in which to create the job.</param> private void CreateJobIfNotExist(BatchClient batchClient, string jobId, string poolId) { //if (batchClient.JobOperations.GetJob(jobId) != null) // return; try { Console.WriteLine("Creating job [{0}]...", jobId); CloudJob job = batchClient.JobOperations.CreateJob(); job.Id = $"{JobName}"; job.PoolInformation = new PoolInformation { PoolId = poolId }; job.Commit(); } catch (BatchException be) { // Accept the specific error code JobExists as that is expected if the job already exists if (be.RequestInformation?.BatchError?.Code == BatchErrorCodeStrings.JobExists) { Console.WriteLine("The job {0} already existed when we tried to create it", jobId); } else { throw; // Any other exception is unexpected } } } /// <summary> /// /// </summary>Creates tasks to process each of the specified input files, and submits them /// to the specified job for execution. /// <param name="batchClient">A BatchClient object.</param> /// <param name="jobId">ID of the job to which the tasks are added.</param> /// <param name="inputFiles">A collection of ResourceFile objects representing the input file /// to be processed by the tasks executed on the compute nodes.</param> /// <param name="outputContainerSasUrl">The shared access signature URL for the Azure /// Storagecontainer that will hold the output files that the tasks create.</param> /// <returns>A collection of the submitted cloud tasks.</returns> private List<CloudTask> AddTasks(BatchClient batchClient,EncodeJob job, string outputContainerSasUrl) { // Create a collection to hold the tasks added to the job: List<CloudTask> tasks = new List<CloudTask>(); // Assign a task ID for each iteration var taskId = String.Format("Task{0}", Guid.NewGuid()); // Define task command line to convert the video format from MP4 to MP3 using ffmpeg. // Note that ffmpeg syntax specifies the format as the file extension of the input file // and the output file respectively. In this case inputs are MP4. string appPath = String.Format("%AZ_BATCH_APP_PACKAGE_{0}#{1}%", appPackageId, appPackageVersion); string inputMediaFile = job.Name; string outputMediaFile = job.OutputName; string taskCommandLine = String.Format("cmd /c {0}\\ffmpeg.exe {1}", appPath, job.Command); // Create a cloud task (with the task ID and command line) and add it to the task list CloudTask task = new CloudTask(taskId, taskCommandLine); task.ApplicationPackageReferences = new List<ApplicationPackageReference> { new ApplicationPackageReference { ApplicationId = appPackageId, Version = appPackageVersion } }; task.ResourceFiles = new List<ResourceFile>(); task.ResourceFiles.Add(new ResourceFile(Uri.EscapeUriString(job.InputUri.ToString()), inputMediaFile)); // Task output file will be uploaded to the output container in Storage. List<OutputFile> outputFileList = new List<OutputFile>(); OutputFileBlobContainerDestination outputContainer = new OutputFileBlobContainerDestination(outputContainerSasUrl,$"{job.Name}_{job.id}/{job.OutputName}"); OutputFile outputFile = new OutputFile(outputMediaFile, new OutputFileDestination(outputContainer), new OutputFileUploadOptions(OutputFileUploadCondition.TaskSuccess)); outputFileList.Add(outputFile); task.OutputFiles = outputFileList; tasks.Add(task); // Call BatchClient.JobOperations.AddTask() to add the tasks as a collection rather than making a // separate call for each. Bulk task submission helps to ensure efficient underlying API // calls to the Batch service. batchClient.JobOperations.AddTask($"{JobName}", tasks); return tasks; } private CloudBlobClient initBlobClient() { CloudStorageAccount storageOutputAccount = CloudStorageAccount.Parse(outputStorageConnection); // Create a blob client for interacting with the blob service. var blobOutputClient = storageOutputAccount.CreateCloudBlobClient(); return blobOutputClient; //blobOutputContainer = blobOutputClient.GetContainerReference(blobOutputContainerName); //blobOutputContainer.CreateIfNotExists(); } public void QueueTask(EncodeJob job) { BatchSharedKeyCredentials sharedKeyCredentials = new BatchSharedKeyCredentials(batchAccountUrl, batchAccount, batchKey); var blobClient = initBlobClient(); var outputContainerSasUrl = GetContainerSasUrl(blobClient, outputContainerName, SharedAccessBlobPermissions.Write); using (BatchClient batchClient = BatchClient.Open(sharedKeyCredentials)) { // Create the Batch pool, which contains the compute nodes that execute the tasks. CreatePoolIfNotExist(batchClient, PoolId); // Create the job that runs the tasks. CreateJobIfNotExist(batchClient, $"{JobName}", PoolId); // Create a collection of tasks and add them to the Batch job. // Provide a shared access signature for the tasks so that they can upload their output // to the Storage container. AddTasks(batchClient,job,outputContainerSasUrl); } } public async Task<Tuple<string,int>> GetStatus() { BatchSharedKeyCredentials sharedKeyCredentials = new BatchSharedKeyCredentials(batchAccountUrl, batchAccount, batchKey); string result = "正在獲取任務信息..."; int total = 0; using (BatchClient batchClient = BatchClient.Open(sharedKeyCredentials)) { var counts =await batchClient.JobOperations.GetJobTaskCountsAsync(JobName); total = counts.Active + counts.Running + counts.Completed; result = $"總任務:{total},等待的任務:{counts.Active},運行中的任務:{counts.Running},成功的任務:{counts.Succeeded},失敗的任務:{counts.Failed}"; } return new Tuple<string,int>(result,total); } } }
因爲Azure Functions的最大Timeout時間爲10分鐘,當執行一些大型的文件轉換時若是是同步執行每每會致使超時錯誤,因此咱們須要在調用完Batch的任務後便可返回,讓Batch Task後臺執行。爲了監控這些Task的完成情況,咱們須要構建一個定時的Functions來檢查任務狀態。而後將獲取到的狀態信息寫到output Blob Container的status.html中就行了ide
using System; using System.Threading.Tasks; using Microsoft.Azure.WebJobs; using Microsoft.Azure.WebJobs.Host; using Microsoft.WindowsAzure.Storage; using Microsoft.WindowsAzure.Storage.Blob; namespace MS.CSU.mp3encoder { /// <summary> /// 用於更新任務處理狀態 /// </summary> public static class StatusUpdate { static int lastTotal=0; static DateTime lastSubmitTime; static string targetStorageConnection = Environment.GetEnvironmentVariable("targetStorageConnection"); static CloudBlobClient blobOutputClient; static string blobOutputContainerName = Environment.GetEnvironmentVariable("outputContainer") ?? "output"; static CloudBlobContainer blobOutputContainer; [FunctionName("StatusUpdate")] public async static Task Run([TimerTrigger("0 */5 * * * *")]TimerInfo myTimer, TraceWriter log) { string strStatus = ""; int jobCount = 0; try { AzureBatch batch = new AzureBatch(targetStorageConnection); var result=await batch.GetStatus(); strStatus = result.Item1; jobCount = result.Item2 - lastTotal; if (lastTotal != result.Item2) { lastTotal = result.Item2; lastSubmitTime = DateTime.Now; } } catch (Exception err) { strStatus = Uri.EscapeDataString(err.ToString()); }; initBlobContainer(); var statusBlob = blobOutputContainer.GetBlockBlobReference("status.html"); string htmlStatus =$@"<html> <head> <meta http-equiv=""refresh"" content=""5""> < meta charset=""utf-8""> </head> <body> <h1>{strStatus}</h1><br/> <h1>最後更新 :{DateTime.Now.AddHours(8)}</h1> <h1>上次任務提交時間:{lastSubmitTime.AddHours(8)}<h1> <h2>上次任務最後五分鐘內提交了{jobCount}<h2> </body> </html>"; await statusBlob.UploadTextAsync(htmlStatus); } private static void initBlobContainer() { CloudStorageAccount storageOutputAccount = CloudStorageAccount.Parse(targetStorageConnection); // Create a blob client for interacting with the blob service. blobOutputClient = storageOutputAccount.CreateCloudBlobClient(); blobOutputContainer = blobOutputClient.GetContainerReference(blobOutputContainerName); blobOutputContainer.CreateIfNotExists(); } } }
Azure Blob storage bindings for Azure Functions
Timer trigger for Azure Functions