2

I am trying to find out about the performance difference between normal multithreading and multithreading using executor (to maintain a thread pool).

The below are code examples for both.

Without Executor Code (with multithreading):

import java.lang.management.ManagementFactory;
import java.lang.management.MemoryPoolMXBean;
import java.lang.management.MemoryUsage;
import java.lang.management.ThreadMXBean;
import java.util.List;

public class Demo1 {
public static void main(String arg[]) {
    Demo1 demo = new Demo1();
    Thread t5  = new Thread(new Runnable() {
       public void run() {
              int count=0;
              // Thread.State;
              // System.out.println("ClientMsgReceiver started-----");
              Demo1.ChildDemo  obj = new Demo1.ChildDemo();
              while(true) {

                // System.out.println("Threadcount is"+Thread);
                // System.out.println("count is"+(count++));
                Thread t=new Thread(obj);
                t.start();
                ThreadMXBean tb = ManagementFactory.getThreadMXBean();
                List<MemoryPoolMXBean> pools = ManagementFactory.getMemoryPoolMXBeans();
                for (MemoryPoolMXBean pool : pools) {
                   MemoryUsage peak = pool.getPeakUsage();
                   System.out.format("Peak %s memory used: %,d%n",
                             pool.getName(), peak.getUsed());
                   System.out.format("Peak %s memory reserved: %,d%n",
                             pool.getName(), peak.getCommitted());
                } 

                System.out.println("Current Thread Count"+ tb.getThreadCount());
                System.out.println("Peak Thread Count"+ tb.getPeakThreadCount());
                System.out.println("Current_Thread_Cpu_Time " 
                         + tb.getCurrentThreadCpuTime());
                System.out.println("Daemon Thread Count" +tb.getDaemonThreadCount());
       }
       // ChatLogin = new ChatLogin();
     }
  });
  t5.start();
}

static class ChildDemo implements Runnable {
   public void run() {
        try {
        //  System.out.println("Thread Started with custom Run method");
            Thread.sleep(100000);
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        finally {
            System.out.println("A" +Thread.activeCount());
        }
    }
  }
}

With executor (multithreading):

import java.lang.management.ManagementFactory;
import java.lang.management.MemoryPoolMXBean;
import java.lang.management.MemoryUsage;
import java.lang.management.ThreadMXBean;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class Executor_Demo {
public static void main(String arg[]) {
   BlockingQueue<Runnable> queue = new ArrayBlockingQueue<Runnable>(10);
   ThreadPoolExecutor executor   = new ThreadPoolExecutor(
          10, 100, 10, TimeUnit.MICROSECONDS, queue);
   Executor_Demo demo = new Executor_Demo();

   executor.execute(new Runnable() {
       public void run() {
          int count=0;
          // System.out.println("ClientMsgReceiver started-----");
          Executor_Demo.Demo demo2 = new Executor_Demo.Demo();
          BlockingQueue<Runnable> queue1 = new ArrayBlockingQueue<Runnable>(1000);
          ThreadPoolExecutor executor1   = new ThreadPoolExecutor(
                  1000, 10000, 10, TimeUnit.MICROSECONDS, queue1);

          while(true) {
             // System.out.println("Threadcount is"+Thread);
             // System.out.println("count is"+(count++));
             Runnable command= new Demo();
             // executor1.execute(command);
             executor1.submit(command);         
             // Thread t=new Thread(demo2);
             // t.start();
             ThreadMXBean tb = ManagementFactory.getThreadMXBean();
             /* try {
                  executor1.awaitTermination(100, TimeUnit.MICROSECONDS);
                } catch (InterruptedException e) {
                   // TODO Auto-generated catch block
                   e.printStackTrace();
                } */
              List<MemoryPoolMXBean> pools = ManagementFactory.getMemoryPoolMXBeans();
              for (MemoryPoolMXBean pool : pools) {
                 MemoryUsage peak = pool.getPeakUsage();
                 System.out.format("Peak %s memory used: %,d%n",
                          pool.getName(), peak.getUsed());
                 System.out.format("Peak %s memory reserved: %,d%n",
                          pool.getName(), peak.getCommitted());
          }
              System.out.println("daemon threads"+tb.getDaemonThreadCount());
              System.out.println("All threads"+tb.getAllThreadIds());
              System.out.println("current thread CPU time "
                       + tb.getCurrentThreadCpuTime());
              System.out.println("current thread user time "
                       + tb.getCurrentThreadUserTime());
              System.out.println("Total started thread count " 
                       + tb.getTotalStartedThreadCount());
              System.out.println("Current Thread Count"+ tb.getThreadCount());
              System.out.println("Peak Thread Count"+ tb.getPeakThreadCount());
              System.out.println("Current_Thread_Cpu_Time " 
                       + tb.getCurrentThreadCpuTime());
              System.out.println("Daemon Thread Count"
                       + tb.getDaemonThreadCount());
              // executor1.shutdown();  
            }
            //ChatLogin = new ChatLogin();
          }
     });
     executor.shutdown();
}

static class Demo implements Runnable {
    public void run() {
      try {
        // System.out.println("Thread Started with custom Run method");
        Thread.sleep(100000);
      } catch (InterruptedException e) {
          // TODO Auto-generated catch block
          e.printStackTrace();
      }
      finally {
         System.out.println("A" +Thread.activeCount());
      }
   }
  }
}

Sample output Output

When I run both programs, it turns out the executor is more expensive than normal multithreading. why is this so?

And given this, what is the use of executor exactly? We use the executor to manage thread pools.

I would have expected the executor to give better results than normal multithreading.

Basically I'm doing this as I need to handle millions of clients using socket programming with multithreading.

Any suggestions will be helpful.

4

2 回答 2

2

每个线程消耗堆栈内存，从 256K 到 1M。您可以手动设置堆栈大小，但将其设置为 128K 以下是危险的。因此，如果您有 2G 内存并且有能力将 1/2 用于线程，那么您将拥有不超过 8K 的线程。如果这对您来说没问题，请使用普通的多线程（每个 Runnable 都有自己的堆栈）。如果您不愿意或不能为每个 Runnable 花费这么多内存，请使用 Executor。将线程池大小设置为处理器的数量 (Runtime.availableProcessors())，或多几倍。出现的主要问题是，您不能使 Thread.sleep() 或以其他方式阻塞您可运行的线程（例如，等待用户响应），因为这种阻塞有效地将线程排除在服务之外。结果，如果您使用大小有限的线程池，就会发生所谓的“线程饥饿”，这实际上是一个僵局。如果您的线程池大小不受限制，那么您将退回到正常的多线程处理并很快耗尽内存。

解决方法是使用异步操作，即使用您的回调设置一些请求，然后退出 run() 方法。然后回调应该使用 Executor.execute(Runnable) 开始执行一些 Runnable 对象（可能相同），或者它可以执行方法 runnable.run() 本身。

Java 7 (nio2) 中现在存在异步输入输出操作，但我未能使其服务于数百个网络连接。为了服务网络连接，可以使用异步网络库（例如 Apache Netty）。

组织回调和执行可运行文件可能需要复杂的同步。为了让生活更轻松，请考虑使用 Actor 模型 (http://en.wikipedia.org/wiki/Actor_model)，其中 Actor 是每次输入消息到达时执行的 Runnable。存在许多 Java Actor 库（例如https://github.com/rfqu/df4j）。

于 2012-03-07T08:46:02.427 回答
2

为了了解某些东西是如何扩展的，我会尝试将监控成本​​保持在最低水平，并将小数字与大数字进行比较。

public class Executor_Demo {
    public static void main(String... arg) throws ExecutionException, InterruptedException {
        int nThreads = 5100;
        ExecutorService executor = Executors.newFixedThreadPool(nThreads, new DaemonThreadFactory());

        List<Future<Results>> futures = new ArrayList<Future<Results>>();
        for (int i = 0; i < nThreads; i++) {
            futures.add(executor.submit(new BackgroundCallable()));
        }
        Results result = new Results();
        for (Future<Results> future : futures) {
            result.merge(future.get());
        }
        executor.shutdown();

        result.print(System.out);

    }

    static class Results {
        private long cpuTime;
        private long userTime;

        Results() {
            final ThreadMXBean tb = ManagementFactory.getThreadMXBean();
            cpuTime = tb.getCurrentThreadCpuTime();
            userTime = tb.getCurrentThreadUserTime();
        }


        public void merge(Results results) {
            cpuTime += results.cpuTime;
            userTime += results.userTime;
        }

        public void print(PrintStream out) {
            ThreadMXBean tb = ManagementFactory.getThreadMXBean();

            List<MemoryPoolMXBean> pools = ManagementFactory.getMemoryPoolMXBeans();
            for (int i = 0, poolsSize = pools.size(); i < poolsSize; i++) {
                MemoryPoolMXBean pool = pools.get(i);
                MemoryUsage peak = pool.getPeakUsage();
                out.format("Peak %s memory used:\t%,d%n", pool.getName(), peak.getUsed());
                out.format("Peak %s memory reserved:\t%,d%n", pool.getName(), peak.getCommitted());
            }

            out.println("Total thread CPU time\t" + cpuTime);
            out.println("Total thread user time\t" + userTime);
            out.println("Total started thread count\t" + tb.getTotalStartedThreadCount());
            out.println("Current Thread Count\t" + tb.getThreadCount());
            out.println("Peak Thread Count\t" + tb.getPeakThreadCount());
            out.println("Daemon Thread Count\t" + tb.getDaemonThreadCount());
        }
    }

    static class DaemonThreadFactory implements ThreadFactory {
        @Override
        public Thread newThread(Runnable r) {
            Thread t = new Thread(r);
            t.setDaemon(true);
            return t;
        }
    }

    static class BackgroundCallable implements Callable<Results> {
        @Override
        public Results call() throws Exception {
            Thread.sleep(100);
            return new Results();
        }
    }
}

测试时-XX:MaxNewSize=64m（这限制了临时内存空间的大小会增加）

100 threads
Peak Code Cache memory used:    386,880
Peak Code Cache memory reserved:    2,555,904
Peak PS Eden Space memory used: 41,280,984
Peak PS Eden Space memory reserved: 50,331,648
Peak PS Survivor Space memory used: 0
Peak PS Survivor Space memory reserved: 8,388,608
Peak PS Old Gen memory used:    0
Peak PS Old Gen memory reserved:    192,675,840
Peak PS Perm Gen memory used:   3,719,616
Peak PS Perm Gen memory reserved:   21,757,952
Total thread CPU time   20000000
Total thread user time  20000000
Total started thread count  105
Current Thread Count    93
Peak Thread Count   105
Daemon Thread Count 92

5100 threads
Peak Code Cache memory used:    425,728
Peak Code Cache memory reserved:    2,555,904
Peak PS Eden Space memory used: 59,244,544
Peak PS Eden Space memory reserved: 59,244,544
Peak PS Survivor Space memory used: 2,949,152
Peak PS Survivor Space memory reserved: 8,388,608
Peak PS Old Gen memory used:    3,076,400
Peak PS Old Gen memory reserved:    192,675,840
Peak PS Perm Gen memory used:   3,787,096
Peak PS Perm Gen memory reserved:   21,757,952
Total thread CPU time   810000000
Total thread user time  150000000
Total started thread count  5105
Current Thread Count    5105
Peak Thread Count   5105
Daemon Thread Count 5104

主要的增加是使用的老一代增加了 ~ 3 MB 或每个线程约 6 KB。以及每个线程使用 956 毫秒或大约 0.2 毫秒的 CPU。

在您的第一个示例中，您正在创建一个线程，在第二个示例中您正在创建 1000。

您正在执行的输出似乎是大部分工作，并且在第二种情况下您的输出比第一种要多得多。

您需要确保您的测试和监控比您想要监控/测量的要轻得多。

于 2012-03-07T08:02:34.183 回答