建设银行网站logo,加强人社局网站建设,购物网站线下推广办法,开发公司总工程师职责背景我司在很久之前#xff0c;一位很久之前的同事写过一个文档转图片的服务#xff0c;具体业务如下#xff1a;1. 用户在客户端上传文档#xff0c;可以是ppt#xff0c;word#xff0c;pdf 等格式#xff0c;用户上传完成可以在客户端预览上传的文档#xff0c;预览… 背景我司在很久之前一位很久之前的同事写过一个文档转图片的服务具体业务如下1. 用户在客户端上传文档可以是pptwordpdf 等格式用户上传完成可以在客户端预览上传的文档预览的时候采用的是图片形式不要和我说用别的方式预览现在已经来不及了2. 当用户把文档上传到云端之后阿里云把文档相关的信息记录在数据库然后等待转码完成3. 服务器有一个转码服务其实就是一个windows service不停的在轮训待转码的数据如果有待转码的数据则从数据库取出来然后根据文档的网络地址下载到本地进行转码转成多张图片4. 当文档转码完毕把转码出来的图片上传到云端并把云端图片的信息记录到数据库5. 客户端有预览需求的时候根据数据库来判断有没有转码成功如果成功则获取数据来显示。文档预览的整体过程如以上所说老的转码服务现在什么问题呢1. 由于一个文档同时只能被一个线程进行转码操作所以老的服务采用了把待转码数据划分管道的思想一共有六个管道映射到数据库大体就是 Id》管道ID 这个样子。2. 一个控制台程序根据配置文件信息读取某一个管道待转码的文档然后单线程进行转码操作3. 一共有六个管道所以服务器上起了六个cmd的黑窗口......4. 有的时候个别文档由于格式问题或者其他问题 转码过程中会卡住具体的表现为停止了转码操作。5. 如果程序卡住了需要运维人员重新启动转码cmd窗口这种维护比较蛋疼后来机缘巧合这个程序的维护落到的菜菜头上维护了一周左右大约重启了10多次终于忍受不了了重新搞一个吧。仔细分析过后刨除实际文档转码的核心操作之外整个转码流程其实还有很多注意点1. 需要保证转码服务不被卡住如果和以前一样就没有必要重新设计了2. 尽量避免开多个进程的方式其实在这个业务场景下多个进程和多个线程作用是一致的。3. 每个文档只能被转码一次如果一个文档被转码多次不仅浪费了服务器资源而且还有可能会有数据不一致的情况发生4. 转码失败的文档需要有一定次数的重试因为一次失败不代表第二次失败所以一定要给失败的文档再次被操作的机会5. 因为程序不停的把文档转码成本地图片所以需要保证这些文件在转码完成在服务器上删除不然的话时间长了会生成很多无用的文件说了这么多其实需要注意的点还是很多的。以整个的转码流程来说本质上是一个任务池的生产和消费问题任务池中的任务就是待转码的文档生产者不停的把待转码文档丢进任务池消费者不停的把任务池中文档转码完成。线程池这很显然和线程池很类似菜菜之前就写过一个线程池的文章有兴趣的同学可以去翻翻历史。今天我们就以这个线程池来解决这个转码问题。线程池的本质是初始化一定数目的线程不停的执行任务。 //线程池定义 public class LXThreadPool:IDisposable{bool PoolEnable true; //线程池是否可用 ListThread ThreadContainer null; //线程的容器ConcurrentQueueActionData JobContainer null; //任务的容器int _maxJobNumber; //线程池最大job容量ConcurrentDictionarystring, DateTime JobIdList new ConcurrentDictionarystring, DateTime(); //job的副本用于排除某个job 是否在运行中public LXThreadPool(int threadNumber,int maxJobNumber1000){if(threadNumber0 || maxJobNumber 0){throw new Exception(线程池初始化失败);}_maxJobNumber maxJobNumber;ThreadContainer new ListThread(threadNumber);JobContainer new ConcurrentQueueActionData();for (int i 0; i threadNumber; i){var t new Thread(RunJob);t.Name $转码线程{i};ThreadContainer.Add(t);t.Start();}//清除超时任务的线程var tTimeOutJob new Thread(CheckTimeOutJob);tTimeOutJob.Name $清理超时任务线程;tTimeOutJob.Start();}//往线程池添加一个线程,返回线程池的新线程数public int AddThread(int number1){if(!PoolEnable || ThreadContainernull || !ThreadContainer.Any() || JobContainernull|| !JobContainer.Any()){return 0;}while (number 0){var t new Thread(RunJob);ThreadContainer.Add(t);t.Start();number - number;}return ThreadContainer?.Count ?? 0;}//向线程池添加一个任务,返回0添加任务失败 1成功public int AddTask(Actionobject job, object obj,string actionId, ActionException errorCallBack null){if (JobContainer ! null){if(JobContainer.Count _maxJobNumber){return 0;}//首先排除10分钟还没转完的var timeoOutJobList JobIdList.Where(s s.Value.AddMinutes(10) DateTime.Now);if(timeoOutJobList!null timeoOutJobList.Any()){foreach (var timeoutJob in timeoOutJobList){JobIdList.TryRemove(timeoutJob.Key,out DateTime v);}}if (!JobIdList.Any(s s.Key actionId)){if(JobIdList.TryAdd(actionId, DateTime.Now)){JobContainer.Enqueue(new ActionData { Job job, Data obj, ActionId actionId, ErrorCallBack errorCallBack });return 1;}else{return 101;}}else{return 100;} }return 0;} private void RunJob(){while (JobContainer ! null PoolEnable){//任务列表取任务ActionData job null;JobContainer?.TryDequeue(out job);if (job null){//如果没有任务则休眠Thread.Sleep(20);continue;}try{//执行任务job.Job.Invoke(job.Data);}catch (Exception error){//异常回调if (job ! null job.ErrorCallBack!null){job?.ErrorCallBack(error);}}finally{if (!JobIdList.TryRemove(job.ActionId,out DateTime v)){}}}}//终止线程池public void Dispose(){PoolEnable false;JobContainer null;if (ThreadContainer ! null){foreach (var t in ThreadContainer){//强制线程退出并不好会有异常t.Join();}ThreadContainer null;}}//清理超时的任务private void CheckTimeOutJob(){//首先排除10分钟还没转完的var timeoOutJobList JobIdList.Where(s s.Value.AddMinutes(10) DateTime.Now);if (timeoOutJobList ! null timeoOutJobList.Any()){foreach (var timeoutJob in timeoOutJobList){JobIdList.TryRemove(timeoutJob.Key, out DateTime v);}}System.Threading.Thread.Sleep(60000);}}public class ActionData{//任务的id用于排重public string ActionId { get; set; }//执行任务的参数public object Data { get; set; }//执行的任务public Actionobject Job { get; set; }//发生异常时候的回调方法public ActionException ErrorCallBack { get; set; }}
以上就是一个线程池的具体实现和具体的业务无关完全可以用于任何适用于线程池的场景其中有一个注意点我新加了任务的标示主要用于排除重复的任务被投放多次只排除正在运行中的任务。当然代码不是最优的有需要的同学可以自己去优化使用线程池接下来我们利用以上的线程池来完成我们的文档转码任务首先我们启动的时候初始化一个线程池,并启动一个独立线程来不停的往线程池来输送任务顺便起了一个监控线程去监视发送任务的线程string lastResId null;string lastErrorResId null;Dictionarystring, int ResErrNumber new Dictionarystring, int(); //转码失败的资源重试次数int MaxErrNumber 5;//最多转码错误的资源10次Thread tPutJoj null;LXThreadPool pool new LXThreadPool(4,100);public void OnStart(){//初始化一个线程发送转码任务tPutJoj new Thread(PutJob);tPutJoj.IsBackground true;tPutJoj.Start();//初始化 监控线程var tMonitor new Thread(MonitorPutJob);tMonitor.IsBackground true;tMonitor.Start();}//监视发放job的线程private void MonitorPutJob(){while (true){if(tPutJoj null|| !tPutJoj.IsAlive){Log.Error($发送转码任务线程停止);tPutJoj new Thread(PutJob);tPutJoj.Start();Log.Error($发送转码任务线程重新初始化并启动);}System.Threading.Thread.Sleep(5000);}}private void PutJob(){ while (true){try{//先搜索等待转码的var fileList DocResourceRegisterProxy.GetFileList(new int[] { (int)FileToImgStateEnum.Wait }, 30, lastResId);Log.Error($拉取待转码记录总数lastResId:{lastResId},结果{fileList?.Count() ?? 0});if (fileList null || !fileList.Any()){lastResId null;Log.Error($待转码数量为0开始拉取转码失败记录重新转码);//如果无待转则把出错的 尝试fileList DocResourceRegisterProxy.GetFileList(new int[] { (int)FileToImgStateEnum.Error, (int)FileToImgStateEnum.TimeOut, (int)FileToImgStateEnum.Fail }, 1, lastErrorResId);if (fileList null || !fileList.Any()){lastErrorResId null;}else{// Log.Error($开始转码失败记录{JsonConvert.SerializeObject(fileList)});ListDocResourceRegister errFilter new ListDocResourceRegister();foreach (var errRes in fileList){if (ResErrNumber.TryGetValue(errRes.res_id, out int number)){if (number MaxErrNumber){Log.Error($资源:{errRes.res_id} 转了{MaxErrNumber}次不成功放弃);continue;}else{errFilter.Add(errRes);ResErrNumber[errRes.res_id] number 1;}}else{ResErrNumber.Add(errRes.res_id, 1);errFilter.Add(errRes);}}fileList errFilter;if (fileList.Any()){lastErrorResId fileList.Select(s s.res_id).Max();}}}else{lastResId fileList.Select(s s.res_id).Max();}if (fileList ! null fileList.Any()){foreach (var file in fileList){//如果 任务投放线程池失败则等待一面继续投放int poolRet 0;while (poolRet 0){poolRet pool.AddTask(s {AliFileService.ConvertToImg(file.res_id $.{file.res_ext}, FileToImgFac.Instance(file.res_ext));}, file, file.res_id);if (poolRet 0 || poolRet 1){Log.Error($发放转码任务失败线程池返回结果{poolRet});System.Threading.Thread.Sleep(1000);}}}}//每一秒去数据库取一次数据System.Threading.Thread.Sleep(3000);}catch{continue;}}}
以上就是发放任务线程池执行任务的所有代码由于具体的转码代码涉及到隐私这里不在提供如果有需要可以私下找菜菜索要虽然我深知还有更优的方式但是我觉得线程池这样的思想可能会对部分人有帮助其中任务超时的核心代码如下(采用了polly插件)var policy Policy.Timeout(TimeSpan.FromSeconds(this.TimeOut), onTimeout: (context, timespan, task) {ret.StateEnum.FileToImgStateEnum.TimeOut; });policy.Execute(s{.....});
把你的更优方案写在留言区吧2020年大家越来越好●程序员修神之路--打通Docker镜像发布容器运行流程●程序员修神之路--容器技术为什么会这么流行(记得去抽奖)●程序员修神之路--kubernetes是微服务发展的必然产物●程序员过关斩将--要想获取我的用户信息就得按照规矩来●程序员过关斩将--更加优雅的Token认证方式JWT●程序员过关斩将--cookie和session的关系其实很简单●程序员修神之路--用NOSql给高并发系统加速●程序员修神之路--高并发系统设计负载均衡架构●程序员修神之路--做好分库分表其实很难之一继续送书●程序员修神之路--做好分库分表其实很难之二送书继续●程序员过关斩将--你为什么还在用存储过程●程序员过关斩将--小小的分页引发的加班血案●程序员修神之路--问世间异步为何物●程序员修神之路--提高网站的吞吐量????
