多線程處理大量文件

時間 2019-12-06

標籤多線程處理大量文件欄目 Java 简体版

原文原文鏈接

上週作了一個多線程處理大量文件的功能一是記錄二是分享三是請博友指出不足更多的瞭解多線程。html

1.任務：將大量(大約5G)一目錄下有日期規則命名的html文件按照年月日三個層次目錄存放，目的是爲了提升文件檢索效率。多線程

2.具體實現：開啓10個線程將文件拷貝到目標文件夾；不符合要求的文件拷貝到別處；記錄錯誤信息和不符合要求的信息；循環判斷狀態執行完畢給出提示。spa

3.開始設想和後來出現問題：線程

開了10個線程處理全部文件，若是一文件已在目標文件下存在則不處理，可是線程間幾乎是同時進行的這樣就會報錯"該文件已被另外一線程佔用"。這樣處理是不行的因而就改善了。讓每一個線程按順序處理固定條數的文件，則每一個線程不會處理處理同個文件。code

4.代碼實現：orm

　聲明變量htm

    /    /App.config 配置路徑
        //源文件路徑
        string sourcePath = ConfigurationManager.AppSettings["sourcePath"];
        //目標路徑
        string toPath = ConfigurationManager.AppSettings["toPath"];
        //錯誤文件存放路徑
        string errorLogPath = "\\log";
        //文件總數
        int totalFileCount;
        //複製文件數
        private int operaFileCount;
        //未複製文件數
        int notCopyFileCount;
        //文件名稱過長文件數
        int longNameFileCount;
        //線程數
        int threadCount = 10;

將全部文件名稱進行分批處理 分頁方法

        /// <summary>
        /// 將全部文件名稱進行分頁
        /// </summary>
        /// <param name="PageSize">每頁條數</param>
        /// <param name="CurPage">當前頁</param>
        /// <param name="objs">集合</param>
        /// <returns></returns>
        public static List<string> QueryByPage(int PageSize, int CurPage, List<string> objs)
        {
            return objs.Take(PageSize * CurPage).Skip(PageSize * (CurPage - 1)).ToList();
        }
      
        public class Page
        {
            public int pageIndex { get; set; }
            public int pageCount { get; set; }
            public int pageSize { get; set; }
            public List<string> list { get; set; }
        }

拷貝文件的方法blog

        #region DoWork方法
        private string errorFieName;
        public void DoWork(object obj)
        {
            Object locker = new object();
            Page page = (Page)obj;
            Console.WriteLine(ThreadName());
            int PageSize = page.pageSize;
            int CurPage = page.pageIndex;
            var grouList = QueryByPage(PageSize, CurPage, page.list);
            foreach (string file in grouList)
            {
                string fileName = Path.GetFileName(file);
                errorFieName = fileName;

                if (fileName != null && fileName.Length != 18)
                {

                    Console.WriteLine(fileName + "文件名稱不符合要求!");
                    CopyErrorFile(fileName);
                    Write(fileName + "文件名稱不符合要求!");
                    continue;
                }
                //Console.WriteLine(fileName.Length);
                try
                {
                    //截取文件名稱 源文件名稱規則 ABC2014200...html 意思是：2014年第200天 能夠推算出年月日
                    string subYearMonth = fileName.Substring(3, 5);
                    int yearNum = 0;
                    int dayNum = 0;
                    int.TryParse(subYearMonth.Substring(0, 2), out yearNum);
                    int.TryParse(subYearMonth.Substring(2, 3), out dayNum);
                    int.TryParse("20" + yearNum, out yearNum);
                    if (yearNum < 1 || dayNum < 1)
                    {
                        Console.WriteLine(fileName + "文件名稱不符合要求!");
                        CopyErrorFile(fileName);
                        //Write(fileName + "文件名稱不符合要求!");
                        continue;
                    }
                    //聲明日期
                    DateTime date = new DateTime();
                    date = date.AddYears(yearNum).AddYears(-1);
                    date = date.AddDays(dayNum).AddDays(-1);
                    string fullSavePath = string.Format("{0}\\{1}\\{2}\\{3}\\", toPath, yearNum, date.Month, date.Day);
                    if (!Directory.Exists(fullSavePath))
                    {
                        Directory.CreateDirectory(fullSavePath);
                    }
                    lock (fullSavePath)
                    {

                        File.Copy(file, fullSavePath + fileName, true);
                        operaFileCount++;
                        //Write("處理完成：" + fileName);
                    }
                }
                catch (Exception ex)
                {
                    Console.WriteLine("文件名稱：" + errorFieName + "處理錯誤：" + ex.Message);
                    Write(ex.Message);
                    throw new Exception(ex.Message);
                }
            }
        }
        #endregion

循環執行線程ip

        public void CopyFile()
        {
            //開始方法時刪除上次記錄
            DeleteLog();
            List<Thread> threadList = new List<Thread>(); ;
            for (int i = 0; i < threadCount; i++)
            {
                try
                {
                    if (!Directory.Exists(toPath))
                    {
                        Directory.CreateDirectory(toPath);
                    }
                    //string[] fileNames = Directory.GetFileSystemEntries(sourcePath);
                    string[] fileNames = Directory.GetFiles(sourcePath);
                    var fileNameList = fileNames.ToList();
                    totalFileCount=fileNames.Length;
                   
                    //共threadCount個線程 每一個線程執行oneThreadNameCount條
                    int oneThreadNameCount = (int)Math.Ceiling(((double)fileNames.Length) / threadCount);
                    Page page;
                    //當前執行的線程
                    page = new Page { pageIndex = i + 1, pageSize = oneThreadNameCount, list = fileNameList };
                    threadList.Add(new Thread(new ParameterizedThreadStart(DoWork)));
                    threadList[i].Start(page);
                    threadList[i].Name = i + "_thread";
                }

                catch (Exception ex)
                {
                    Console.WriteLine("錯誤信息：" + ex.Message);
                    Write(ex.Message);
                }
            }
            //判斷線程執行狀況
            bool isRanning;
            bool isComplete = false;
            while (!isComplete)
            {
                //2秒判斷一次
                Thread.Sleep(2000);
                isRanning = false;
                foreach (var item in threadList)
                {
                    if (item.ThreadState == ThreadState.Running)
                    {
                        isRanning = true;
                    }
                }
                if (!isRanning) 
                { 
                    isComplete = true;
                    Console.WriteLine();
                    Console.WriteLine("處理結果 共" + totalFileCount+"；已處理："+operaFileCount);
                    Console.WriteLine("不符合要求:" + notCopyFileCount + "(已拷貝到errorfile文件夾)；沒法拷貝名稱過長文件：" + longNameFileCount);
                    Console.WriteLine("請查看文件夾" + toPath);
                }
            }
            Console.ReadKey();
        }

輔助方法get

        #region copy不符合要求文件
        private void CopyErrorFile(string fileName)
        {
            //實際長度超過222報錯，而且沒法拋出異常
            if (fileName.Length > 180)
            {
                longNameFileCount += 1;
                Write(fileName + "名稱過長!");
                return;
            }
            notCopyFileCount += 1;
            string strErrorPath  =toPath+"\\errorfile";
            if(!Directory.Exists(strErrorPath)) Directory.CreateDirectory(strErrorPath);
            File.Copy(sourcePath+"\\"+fileName, toPath+"\\errorfile\\" + fileName, true);
        }
        #endregion

        private void DeleteLog()
        {
            try
            {
                if (!Directory.Exists(toPath + errorLogPath)) return;
                foreach (string var in Directory.GetFiles(toPath + errorLogPath))
                {
                    File.Delete(var);
                }
            }
            catch
            {
                //Directory.CreateDirectory(toPath + errorLogPath);
            }

        }
        private void Write(string message)
        {
            object obj = new object();
            lock (obj)
            {
                try
                {
                    string logPath=toPath+errorLogPath;
                    if (!Directory.Exists(logPath)) Directory.CreateDirectory(logPath);
                    Thread primaryThread = Thread.CurrentThread;
                    //當前線程名稱
                    string threadName = primaryThread.Name;
                    using (TextWriter sw = new StreamWriter(logPath+"\\"+ThreadName()+"_error.txt", true))
                    {
                        sw.WriteLine("錯誤信息：" + message);
                        sw.Dispose();
                    }
                   
                }
                catch (Exception ex)
                {
                    Console.WriteLine("錯誤信息：" + ex.Message);
                }
            }


        }