自定义线程池来实现文档转码
背景
我司在很久之前,一位很久之前的同事写过一个文档转图片的服务,具体业务如下:
用户在客户端上传文档,可以是ppt,word,pdf 等格式,用户上传完成可以在客户端预览上传的文档,预览的时候采用的是图片形式(不要和我说用别的方式预览,现在已经来不及了)
当用户把文档上传到云端之后(阿里云),把文档相关的信息记录在数据库,然后等待转码完成
服务器有一个转码服务(其实就是一个windows service)不停的在轮训待转码的数据,如果有待转码的数据,则从数据库取出来,然后根据文档的网络地址下载到本地进行转码(转成多张图片)
当文档转码完毕,把转码出来的图片上传到云端,并把云端图片的信息记录到数据库
客户端有预览需求的时候,根据数据库来判断有没有转码成功,如果成功,则获取数据来显示。
文档预览的整体过程如以上所说,老的转码服务现在什么问题呢?
由于一个文档同时只能被一个线程进行转码操作,所以老的服务采用了把待转码数据划分管道的思想,一共有六个管道,映射到数据库大体就是 Id=》管道ID 这个样子。
一个控制台程序,根据配置文件信息,读取某一个管道待转码的文档,然后单线程进行转码操作
一共有六个管道,所以服务器上起了六个cmd的黑窗口......
有的时候个别文档由于格式问题或者其他问题 转码过程中会卡住,具体的表现为:停止了转码操作。
如果程序卡住了,需要运维人员重新启动转码cmd窗口(这种维护比较蛋疼)
后来机缘巧合,这个程序的维护落到的菜菜头上,维护了一周左右,大约重启了10多次,终于忍受不了了,重新搞一个吧。仔细分析过后,刨除实际文档转码的核心操作之外,整个转码流程其实还有很多注意点
需要保证转码服务不被卡住,如果和以前一样就没有必要重新设计了
尽量避免开多个进程的方式,其实在这个业务场景下,多个进程和多个线程作用是一致的。
每个文档只能被转码一次,如果一个文档被转码多次,不仅浪费了服务器资源,而且还有可能会有数据不一致的情况发生
转码失败的文档需要有一定次数的重试,因为一次失败不代表第二次失败,所以一定要给失败的文档再次被操作的机会
因为程序不停的把文档转码成本地图片,所以需要保证这些文件在转码完成在服务器上删除,不然的话,时间长了会生成很多无用的文件
说了这么多,其实需要注意的点还是很多的。以整个的转码流程来说,本质上是一个任务池的生产和消费问题,任务池中的任务就是待转码的文档,生产者不停的把待转码文档丢进任务池,消费者不停的把任务池中文档转码完成。
线程池
这很显然和线程池很类似,菜菜之前就写过一个线程池的文章,有兴趣的同学可以去翻翻历史。今天我们就以这个线程池来解决这个转码问题。线程池的本质是初始化一定数目的线程,不停的执行任务。
//线程池定义 public class LXThreadPool:IDisposable { bool PoolEnable = true; //线程池是否可用 List<Thread> ThreadContainer = null; //线程的容器 ConcurrentQueue<ActionData> JobContainer = null; //任务的容器 int _maxJobNumber; //线程池最大job容量 ConcurrentDictionary<string, DateTime> JobIdList = new ConcurrentDictionary<string, DateTime>(); //job的副本,用于排除某个job 是否在运行中 public LXThreadPool(int threadNumber,int maxJobNumber=1000) { if(threadNumber<=0 || maxJobNumber <= 0) { throw new Exception("线程池初始化失败"); } _maxJobNumber = maxJobNumber; ThreadContainer = new List<Thread>(threadNumber); JobContainer = new ConcurrentQueue<ActionData>(); 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 number=1) { if(!PoolEnable || ThreadContainer==null || !ThreadContainer.Any() || JobContainer==null|| !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(Action<object> job, object obj,string actionId, Action<Exception> 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 Action<object> Job { get; set; } //发生异常时候的回调方法 public Action<Exception> ErrorCallBack { get; set; } }
以上就是一个线程池的具体实现,和具体的业务无关,完全可以用于任何适用于线程池的场景,其中有一个注意点,我新加了任务的标示,主要用于排除重复的任务被投放多次(只排除正在运行中的任务)。当然代码不是最优的,有需要的同学可以自己去优化
使用线程池
接下来,我们利用以上的线程池来完成我们的文档转码任务,首先我们启动的时候初始化一个线程池,并启动一个独立线程来不停的往线程池来输送任务,顺便起了一个监控线程去监视发送任务的线程
string lastResId = null; string lastErrorResId = null; Dictionary<string, int> ResErrNumber = new Dictionary<string, 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)}"); List<DocResourceRegister> errFilter = new List<DocResourceRegister>(); 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.State=Enum.FileToImgStateEnum.TimeOut; }); policy.Execute(s=>{ ..... });
把你的更优方案写在留言区吧,2020年大家越来越好
更多精彩文章
版权声明: 本文为 InfoQ 作者【架构师修行之路】的原创文章。
原文链接:【http://xie.infoq.cn/article/ee5c41e1c6276f191c01a0632】。文章转载请联系作者。
架构师修行之路
一个坚持把架构做到极致的技术人 2014.10.19 加入
还未添加个人简介
评论