Java多线程优化方法及使用方式,详解CodeTimer及

2019-12-03 03:55 来源:未知

详解CodeTimerXCode性能测试是本文要介绍的内容,在测试XCode性能的时候,发现每次执行测试程序得到的执行时间差距实在太大,于是采用了老赵的CodeTimer来计算线程时间,后来因为测试程序稍微有点复杂,在使用匿名委托时会有参数的“打包”过程,于是改进了CodeTimer测试功能代码通过实现一个继承自CodeTimer的类来实现,避免每次迭代时参数“打包”的过程。

Java多线程优化方法及使用方式,java多线程优化方式

一、多线程介绍

在编程中,我们不可逃避的会遇到多线程的编程问题,因为在大多数的业务系统中需要并发处理,如果是在并发的场景中,多线程就非常重要了。另外,我们在面试的时候,面试官通常也会问到我们关于多线程的问题,如:如何创建一个线程?我们通常会这么回答,主要有两种方法,第一种:继承Thread类,重写run方法;第二种:实现Runnable接口,重写run方法。那么面试官一定会问这两种方法各自的优缺点在哪,不管怎么样,我们会得出一个结论,那就是使用方式二,因为面向对象提倡少继承,尽量多用组合。

这个时候,我们还可能想到,如果想得到多线程的返回值怎么办呢?根据我们多学到的知识,我们会想到实现Callable接口,重写call方法。那么多线程到底在实际项目中怎么使用呢,他有多少种方式呢?

首先,我们来看一个例子:

图片 1 

这是一种创建多线程的简单方法,很容易理解,在例子中,根据不同的业务场景,我们可以在Thread()里边传入不同的参数实现不同的业务逻辑,但是,这个方法创建多线程暴漏出来的问题就是反复创建线程,而且创建线程后还得销毁,如果对并发场景要求低的情况下,这种方式貌似也可以,但是高并发的场景中,这种方式就不行了,因为创建线程销毁线程是非常耗资源的。所以根据经验,正确的做法是我们使用线程池技术,JDK提供了多种线程池类型供我们选择,具体方式可以查阅jdk的文档。

图片 2 

这里代码我们需要注意的是,传入的参数代表我们配置的线程数,是不是越多越好呢?肯定不是。因为我们在配置线程数的时候要充分考虑服务器的性能,线程配置的多,服务器的性能未必就优。通常,机器完成的计算是由线程数决定的,当线程数到达峰值,就无法在进行计算了。如果是耗CPU的业务逻辑(计算较多),线程数和核数一样就到达峰值了,如果是耗I/O的业务逻辑(操作数据库,文件上传、下载等),线程数越多一定意义上有助于提升性能。

线程数大小的设定又一个公式决定:

Y=N*((a+b)/a),其中,N:CPU核数,a:线程执行时程序的计算时间,b:线程执行时,程序的阻塞时间。有了这个公式后,线程池的线程数配置就会有约束了,我们可以根据机器的实际情况灵活配置。

二、多线程优化及性能比较

最近的项目中用到了所线程技术,在使用过程中遇到了很多的麻烦,趁着热度,整理一下几种多线程框架的性能比较。目前所掌握的大致分三种,第一种:ThreadPool(线程池)+CountDownLatch(程序计数器),第二种:Fork/Join框架,第三种JDK8并行流,下面对这几种方式的多线程处理性能做一下比较总结。

首先,假设一种业务场景,在内存中生成多个文件对象,这里暂定30000,(Thread.sleep(时间))线程睡眠模拟业务处理业务逻辑,来比较这几种方式的多线程处理性能。

1) 单线程

这种方式非常简单,但是程序在处理的过程中非常的耗时,使用的时间会很长,因为每个线程都在等待当前线程执行完才会执行,和多线程没有多少关系,所以效率非常低。

首先创建文件对象,代码如下:

public class FileInfo {
 private String fileName;//文件名
 private String fileType;//文件类型
 private String fileSize;//文件大小
 private String fileMD5;//MD5码
 private String fileVersionNO;//文件版本号
 public FileInfo() {
  super();
 }
 public FileInfo(String fileName, String fileType, String fileSize, String fileMD5, String fileVersionNO) {
  super();
  this.fileName = fileName;
  this.fileType = fileType;
  this.fileSize = fileSize;
  this.fileMD5 = fileMD5;
  this.fileVersionNO = fileVersionNO;
 }
 public String getFileName() {
  return fileName;
 }
 public void setFileName(String fileName) {
  this.fileName = fileName;
 }
 public String getFileType() {
  return fileType;
 }
 public void setFileType(String fileType) {
  this.fileType = fileType;
 }
 public String getFileSize() {
  return fileSize;
 }
 public void setFileSize(String fileSize) {
  this.fileSize = fileSize;
 }
 public String getFileMD5() {
  return fileMD5;
 }
 public void setFileMD5(String fileMD5) {
  this.fileMD5 = fileMD5;
 }
 public String getFileVersionNO() {
  return fileVersionNO;
 }
 public void setFileVersionNO(String fileVersionNO) {
  this.fileVersionNO = fileVersionNO;
 }

接着,模拟业务处理,创建30000个文件对象,线程睡眠1ms,之前设置的1000ms,发现时间很长,整个Eclipse卡掉了,所以将时间改为了1ms。

public class Test {
   private static List<FileInfo> fileList= new ArrayList<FileInfo>();
   public static void main(String[] args) throws InterruptedException {
     createFileInfo();
     long startTime=System.currentTimeMillis();
     for(FileInfo fi:fileList){
       Thread.sleep(1);
     }
     long endTime=System.currentTimeMillis();
     System.out.println("单线程耗时:"+(endTime-startTime)+"ms");
   }
   private static void createFileInfo(){
     for(int i=0;i<30000;i++){
       fileList.add(new FileInfo("身份证正面照","jpg","101522","md5"+i,"1"));
     }
   }
}

测试结果如下:

图片 3 

可以看到,生成30000个文件对象消耗的时间比较长,接近1分钟,效率比较低。

2) ThreadPool (线程池) +CountDownLatch (程序计数器)

顾名思义,CountDownLatch为线程计数器,他的执行过程如下:首先,在主线程中调用await()方法,主线程阻塞,然后,将程序计数器作为参数传递给线程对象,最后,每个线程执行完任务后,调用countDown()方法表示完成任务。countDown()被执行多次后,主线程的await()会失效。实现过程如下:

public class Test2 {
 private static ExecutorService executor=Executors.newFixedThreadPool(100);
 private static CountDownLatch countDownLatch=new CountDownLatch(100);
 private static List<FileInfo> fileList= new ArrayList<FileInfo>();
 private static List<List<FileInfo>> list=new ArrayList<>();
 public static void main(String[] args) throws InterruptedException {
  createFileInfo();
  addList();
  long startTime=System.currentTimeMillis();
  int i=0;
  for(List<FileInfo> fi:list){
   executor.submit(new FileRunnable(countDownLatch,fi,i));
   i++;
  }
  countDownLatch.await();
  long endTime=System.currentTimeMillis();
  executor.shutdown();
  System.out.println(i+"个线程耗时:"+(endTime-startTime)+"ms");
 }
 private static void createFileInfo(){
  for(int i=0;i<30000;i++){
   fileList.add(new FileInfo("身份证正面照","jpg","101522","md5"+i,"1"));
  }
 }
 private static void addList(){
  for(int i=0;i<100;i++){
   list.add(fileList);
  }
 }
}

FileRunnable类:

/**
 * 多线程处理
 * @author wangsj
 *
 * @param <T>
 */
public class FileRunnable<T> implements Runnable {
   private CountDownLatch countDownLatch;
   private List<T> list;
   private int i;
   public FileRunnable(CountDownLatch countDownLatch, List<T> list, int i) {
     super();
     this.countDownLatch = countDownLatch;
     this.list = list;
     this.i = i;
   }
   @Override
   public void run() {
     for(T t:list){
       try {
          Thread.sleep(1);
       } catch (InterruptedException e) {
          e.printStackTrace();
       }
       countDownLatch.countDown();
     }
   }
}

测试结果如下:

图片 4 

3) Fork/Join 框架

Jdk从版本7开始,出现了Fork/join框架,从字面来理解,fork就是拆分,join就是合并,所以,该框架的思想就是。通过fork拆分任务,然后join来合并拆分后各个人物执行完毕后的结果并汇总。比如,我们要计算连续相加的几个数,2+4+5+7=?,我们利用Fork/join框架来怎么完成呢,思想就是拆分子任务,我们可以把这个运算拆分为两个子任务,一个计算2+4,另一个计算5+7,这是Fork的过程,计算完成后,把这两个子任务计算的结果汇总,得到总和,这是join的过程。

Fork/Join框架执行思想:首先,分割任务,使用fork类将大任务分割为若干子任务,这个分割过程需要按照实际情况来定,直到分割出的任务足够小。然后,join类执行任务,分割的子任务在不同的队列里,几个线程分别从队列里获取任务并执行,执行完的结果放到一个单独的队列里,最后,启动线程,队列里拿取结果并合并结果。

使用Fork/Join框架要用到几个类,关于类的使用方式可以参考JDK的API,使用该框架,首先需要继承ForkJoinTask类,通常,只需要继承他的子类RecursiveTask或RecursiveAction即可,RecursiveTask,用于有返回结果的场景,RecursiveAction用于没有返回结果的场景。ForkJoinTask的执行需要用到ForkJoinPool来执行,该类用于维护分割出的子任务添加到不同的任务队列。

下面是实现代码:

public class Test3 {
 private static List<FileInfo> fileList= new ArrayList<FileInfo>();
// private static ForkJoinPool forkJoinPool=new ForkJoinPool(100);
// private static Job<FileInfo> job=new Job<>(fileList.size()/100, fileList);
 public static void main(String[] args) {
  createFileInfo();
  long startTime=System.currentTimeMillis();
  ForkJoinPool forkJoinPool=new ForkJoinPool(100);
  //分割任务
  Job<FileInfo> job=new Job<>(fileList.size()/100, fileList);
  //提交任务返回结果
ForkJoinTask<Integer> fjtResult=forkJoinPool.submit(job);
//阻塞
  while(!job.isDone()){
   System.out.println("任务完成!");
  }
  long endTime=System.currentTimeMillis();
  System.out.println("fork/join框架耗时:"+(endTime-startTime)+"ms");
 }
 private static void createFileInfo(){
  for(int i=0;i<30000;i++){
   fileList.add(new FileInfo("身份证正面照","jpg","101522","md5"+i,"1"));
  }
 }
}
/**
 * 执行任务类
 * @author wangsj
 *
 */
public class Job<T> extends RecursiveTask<Integer> {
 private static final long serialVersionUID = 1L;
 private int count;
 private List<T> jobList;
 public Job(int count, List<T> jobList) {
  super();
  this.count = count;
  this.jobList = jobList;
 }
 /**
  * 执行任务,类似于实现Runnable接口的run方法
  */
 @Override
 protected Integer compute() {
  //拆分任务
  if(jobList.size()<=count){
   executeJob();
   return jobList.size();
  }else{
   //继续创建任务,直到能够分解执行
   List<RecursiveTask<Long>> fork = new LinkedList<RecursiveTask<Long>>();
   //拆分子任务,这里采用二分法
   int countJob=jobList.size()/2;
   List<T> leftList=jobList.subList(0, countJob);
   List<T> rightList=jobList.subList(countJob, jobList.size());
   //分配任务
   Job leftJob=new Job<>(count,leftList);
   Job rightJob=new Job<>(count,rightList);
   //执行任务
   leftJob.fork();
   rightJob.fork();
   return Integer.parseInt(leftJob.join().toString())
     +Integer.parseInt(rightJob.join().toString());
  }
 }
 /**
  * 执行任务方法
  */
 private void executeJob() {
  for(T job:jobList){
   try {
    Thread.sleep(1);
   } catch (InterruptedException e) {
    e.printStackTrace();
   }
  }
 }

测试结果如下:

图片 5 

4) JDK8 并行流

并行流是jdk8的新特性之一,思想就是将一个顺序执行的流变为一个并发的流,通过调用parallel()方法来实现。并行流将一个流分成多个数据块,用不同的线程来处理不同的数据块的流,最后合并每个块数据流的处理结果,类似于Fork/Join框架。

并行流默认使用的是公共线程池ForkJoinPool,他的线程数是使用的默认值,根据机器的核数,我们可以适当调整线程数的大小。线程数的调整通过以下方式来实现。

System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", "100");

以下是代码的实现过程,非常简单:

public class Test4 {
private static List<FileInfo> fileList= new ArrayList<FileInfo>();
public static void main(String[] args) {
//    System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", "100");
   createFileInfo();
   long startTime=System.currentTimeMillis();
   fileList.parallelStream().forEach(e ->{
     try {
        Thread.sleep(1);
     } catch (InterruptedException f) {
        f.printStackTrace();
     }
   });
   long endTime=System.currentTimeMillis();
   System.out.println("jdk8并行流耗时:"+(endTime-startTime)+"ms");
}
private static void createFileInfo(){
   for(int i=0;i<30000;i++){
     fileList.add(new FileInfo("身份证正面照","jpg","101522","md5"+i,"1"));
   }
}
}

下面是测试,第一次没有设置线程池的数量,采用默认,测试结果如下:

图片 6 

我们看到,结果并不是很理想,耗时较长,接下来设置线程池的数量大小,即添加如下代码:

System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", "100");

接着进行测试,结果如下:

图片 7 

这次耗时较小,比较理想。

三、总结

综上几种情况来看,以单线程作为参考,耗时最长的还是原生的Fork/Join框架,这里边尽管配置了线程池的数量,但效果较精确配置了线程池数量的JDK8并行流较差。并行流实现代码简单易懂,不需要我们写多余的for循环,一个parallelStream方法全部搞定,代码量大大的减少了,其实,并行流的底层还是使用的Fork/Join框架,这就要求我们在开发的过程中灵活使用各种技术,分清各种技术的优缺点,从而能够更好的为我们服务。

一、多线程介绍 在编程中,我们不可逃避的会遇到多线程的编程问题,因为在大多数的...

using System;  using System.Collections.Generic;  using System.Text;  using System.Runtime.InteropServices;  using System.Diagnostics;  using System.Threading;  using NewLife.Reflection;  using NewLife.Exceptions;   namespace NewLife.Log  {      /// <summary>     /// 代码性能计时器      /// </summary>     /// <remarks>参考了老赵http://www.cnblogs.com/jeffreyzhao/archive/2009/03/10/codetimer.html)  和eaglethttp://www.cnblogs.com/eaglet/archive/2009/03/10/1407791.html)两位的作品</remarks>     /// <remarks>为了保证性能比较的公平性,采用了多种指标,并使用计时器重写等手段来避免各种不必要的损耗</remarks>     public class CodeTimer      {          #region 静态快速计时          /// <summary>         /// 计时          /// </summary>         /// <param name="times"></param>         /// <param name="action"></param>         /// <returns></returns>         public static CodeTimer Time(Int32 times, Action<Int32> action)          {              CodeTimer timer = new CodeTimer();              timer.Times = times;              timer.Action = action;               timer.TimeOne();              timer.Time();               return timer;          }           /// <summary>         /// 计时,并用控制台输出行          /// </summary>         /// <param name="title"></param>         /// <param name="times"></param>         /// <param name="action"></param>         public static void TimeLine(String title, Int32 times, Action<Int32> action)          {              Console.Write("{0,16}:", title);               CodeTimer timer = new CodeTimer();              timer.Times = times;              timer.Action = action;              timer.ShowProgress = true;               ConsoleColor currentForeColor = Console.ForegroundColor;              Console.ForegroundColor = ConsoleColor.Yellow;               timer.TimeOne();              timer.Time();               Console.WriteLine(timer.ToString());               Console.ForegroundColor = currentForeColor;          }          #endregion           #region PInvoke          [DllImport("kernel32.dll")]          [return: MarshalAs(UnmanagedType.Bool)]          static extern bool QueryThreadCycleTime(IntPtr threadHandle, ref ulong cycleTime);           [DllImport("kernel32.dll")]          static extern IntPtr GetCurrentThread();           [DllImport("kernel32.dll", SetLastError = true)]          static extern bool GetThreadTimes(IntPtr hThread, out long lpCreationTime, out long lpExitTime,                out long lpKernelTime, out long lpUserTime);           static Boolean supportCycle = true;          private static ulong GetCycleCount()          {              //if (Environment.Version.Major < 6) return 0;               if (!supportCycle) return 0;               try              {                  ulong cycleCount = 0;                  QueryThreadCycleTime(GetCurrentThread(), ref cycleCount);                  return cycleCount;              }              catch              {                  supportCycle = false;                  return 0;              }          }           private static long GetCurrentThreadTimes()          {              long l;              long kernelTime, userTimer;              GetThreadTimes(GetCurrentThread(), out l, out l, out kernelTime, out userTimer);              return kernelTime + userTimer;          }          #endregion           #region 私有字段          ulong cpuCycles = 0;          long threadTime = 0;          int[] gen;          #endregion           #region 属性          private Int32 _Times;          /// <summary>次数</summary>         public Int32 Times          {              get { return _Times; }              set { _Times = value; }          }           private Action<Int32> _Action;          /// <summary>迭代方法,如不指定,则使用Time(int index)</summary>         public Action<Int32> Action          {              get { return _Action; }              set { _Action = value; }          }           private Boolean _ShowProgress;          /// <summary>是否显示控制台进度</summary>         public Boolean ShowProgress          {              get { return _ShowProgress; }              set { _ShowProgress = value; }          }           private Int32 _Index;          /// <summary>进度</summary>         public Int32 Index          {              get { return _Index; }              set { _Index = value; }          }           private ulong _CpuCycles;          /// <summary>CPU周期</summary>         public ulong CpuCycles          {              get { return _CpuCycles; }              set { _CpuCycles = value; }          }           private long _ThreadTime;          /// <summary>线程时间,单位是100ns,除以10000转为ms</summary>         public long ThreadTime          {              get { return _ThreadTime; }              set { _ThreadTime = value; }          }           private Int32[] _Gen = new Int32[] { 0, 0, 0 };          /// <summary>GC代数</summary>         public Int32[] Gen          {              get { return _Gen; }              set { _Gen = value; }          }           private TimeSpan _Elapsed;          /// <summary>执行时间</summary>         public TimeSpan Elapsed          {              get { return _Elapsed; }              set { _Elapsed = value; }          }          #endregion           #region 方法          /// <summary>         /// 计时核心方法,处理进程和线程优先级          /// </summary>         public virtual void Time()          {              if (Times <= 0) throw new XException("非法迭代次数!");               // 设定进程、线程优先级,并在完成时还原              ProcessPriorityClass pp = Process.GetCurrentProcess().PriorityClass;              ThreadPriority tp = Thread.CurrentThread.Priority;              try              {                  Process.GetCurrentProcess().PriorityClass = ProcessPriorityClass.High;                  Thread.CurrentThread.Priority = ThreadPriority.Highest;                   StartProgress();                   TimeTrue();              }              finally              {                  StopProgress();                   Thread.CurrentThread.Priority = tp;                  Process.GetCurrentProcess().PriorityClass = pp;              }          }           /// <summary>         /// 真正的计时          /// </summary>         protected virtual void TimeTrue()          {              if (Times <= 0) throw new XException("非法迭代次数!");               // 统计GC代数              GC.Collect(GC.MaxGeneration, GCCollectionMode.Forced);              gen = new Int32[GC.MaxGeneration + 1];              for (Int32 i = 0; i <= GC.MaxGeneration; i++)              {                  gen[i] = GC.CollectionCount(i);              }               Stopwatch watch = new Stopwatch();              watch.Start();              cpuCycles = GetCycleCount();              threadTime = GetCurrentThreadTimes();               // 如果未指定迭代方法,则使用内部的Time              Action<Int32> action = Action;              if (action == null)              {                  action = Time;                   // 初始化                  Init();              }               for (Int32 i = 0; i < Times; i++)              {                  Index = i;                   action(i);              }              if (Action == null)              {                  // 结束                  Finish();              }               CpuCycles = GetCycleCount() - cpuCycles;              ThreadTime = GetCurrentThreadTimes() - threadTime;               watch.Stop();              Elapsed = watch.Elapsed;               // 统计GC代数              List<Int32> list = new List<Int32>();              for (Int32 i = 0; i <= GC.MaxGeneration; i++)              {                  int count = GC.CollectionCount(i) - gen[i];                  list.Add(count);              }              Gen = list.ToArray();          }           /// <summary>         /// 执行一次迭代,预热所有方法          /// </summary>         public void TimeOne()          {              Int32 n = Times;               try              {                  Times = 1;                  Time();              }              finally { Times = n; }          }           /// <summary>         /// 迭代前执行,计算时间          /// </summary>         public virtual void Init() { }           /// <summary>         /// 每一次迭代,计算时间          /// </summary>         /// <param name="index"></param>         public virtual void Time(Int32 index) { }           /// <summary>         /// 迭代后执行,计算时间          /// </summary>         public virtual void Finish() { }          #endregion           #region 进度          Thread thread;           void StartProgress()          {              if (!ShowProgress) return;               // 使用低优先级线程显示进度              thread = new Thread(new ParameterizedThreadStart(Progress));              thread.IsBackground = true;              thread.Priority = ThreadPriority.BelowNormal;              thread.Start();          }           void StopProgress()          {              if (thread != null && thread.IsAlive)              {                  thread.Abort();                  thread.Join(3000);              }          }           void Progress(Object state)          {              Int32 left = Console.CursorLeft;               // 设置光标不可见              Boolean cursorVisible = Console.CursorVisible;              Console.CursorVisible = false;               Stopwatch sw = new Stopwatch();              sw.Start();              while (true)              {                  try                  {                      Int32 i = Index;                      if (i >= Times) break;                       if (i > 0 && sw.Elapsed.TotalMilliseconds > 10)                      {                          Double d = (Double)i / Times;                          Console.Write("{0,7:n0}ms {1:p}", sw.Elapsed.TotalMilliseconds, d);                          Console.CursorLeft = left;                      }                  }                  catch (ThreadAbortException) { break; }                  catch { break; }                   Thread.Sleep(500);              }              sw.Stop();               Console.CursorLeft = left;              Console.CursorVisible = cursorVisible;          }          #endregion           #region 重载          /// <summary>         /// 已重载。输出依次分别是:执行时间、CPU线程时间、时钟周期、GC代数          /// </summary>         /// <returns></returns>         public override string ToString()          {              return String.Format("{0,7:n0}ms {1,7:n0}ms {2,15:n0} {3}/{4}/{5}", Elapsed.TotalMilliseconds,                  ThreadTime / 10000, CpuCycles, Gen[0], Gen[1], Gen[2]);          }          #endregion      }  } 

对于控制台测试项目,另外起了一个线程负责输出进度,不知道这样对测试会有多大影响。

CodeTimer 及 XCode 性能 测试 是本文要介绍的内容,在 测试XCode 性能的时候,发现每次执行 测试 程序得到的执行时间差距实在太大,于...

TAG标签:
版权声明:本文由990888藏宝阁发布于网络应用,转载请注明出处:Java多线程优化方法及使用方式,详解CodeTimer及