PriorityBlockingQueue 和 Executors.newCachedThreadPool()

一、PriorityBlockingQueue里面存储的对象必须是实现Comparable接口。

二、队列经过这个接口的compare方法肯定对象的优先级priority。

规则是：当前和其余对象比较，若是compare方法返回负数，那么在队列里面的优先级就比较高。

下面的测试能够说明这个断言：

查看打印结果，比较take出来的Entity和left的entity，比较他们的priority

public class TestPriorityQueue { 

static Random r=new Random(47); 

public static void main(String args[]){ 

final PriorityBlockingQueue q = new PriorityBlockingQueue(); 

ExecutorService se = Executors.newCachedThreadPool(); 

final int qTime = r.nextInt(100);

//execute producer 
se.execute(new Runnable(){ 
        public void run() { 
        int i=0; 
        while(true){ 
            q.put(new PriorityEntity(r.nextInt(10), i++));  //优先级，索引
            try { 
                TimeUnit.MILLISECONDS.sleep(qTime); 
            } catch (InterruptedException e) {
                e.printStackTrace(); 
            } 
        } 
    } 
}); 

//execute consumer 
se.execute(new Runnable(){ 
    public void run() { 
        while(true){ 
        try { 
            int rTime = r.nextInt(500);

            //打印进队和出队的时间能够看到队列一直在累积
            System.out.println(qTime + "： " rTime + " --take-- "+q.take()+" left: "+q.size()+" --     ["+q.toString()+"]"); 
            try { 
                TimeUnit.MILLISECONDS.sleep(rTime); 
            } catch (InterruptedException e) { 
                e.printStackTrace(); 
            } 
            } catch (InterruptedException e) { 
                e.printStackTrace(); 
            }
        } 
    } 
});

    //try { 
    //    TimeUnit.SECONDS.sleep(5); 
    //} catch (InterruptedException e) {
    //    e.printStackTrace(); 
    //} 
    System.out.println("shutdown"); 
} 

} 

class PriorityEntity implements Comparable<PriorityEntity> {

    private int priority; 
    private int index=0; 

    public PriorityEntity(int _priority,int _index) { 
        this.priority = _priority; 
        this.index=_index; 
    } 

    public String toString(){ 
        return "# [index="+index+" priority="+priority+"]"; 
    } 

//数字小，优先级高 
public int compareTo(PriorityEntity o) { 
    return this.priority > o.priority ? 1 : this.priority < o.priority ? -1 : 0;  
} 

//数字大，优先级高 
// public int compareTo(PriorityTask o) { 
//     return this.priority < o.priority ? 1 : this.priority > o.priority ? -1 : 0;  
// } 
}        

 

 

理解 newCachedThreadPool()

一、建立线程池的根源：

public ThreadPoolExecutor(int corePoolSize, 
    int maximumPoolSize,
    long keepAliveTime,
    TimeUnit unit,
    BlockingQueue<Runnable> workQueue,
    ThreadFactory threadFactory,
    RejectedExecutionHandler handler)

 

二、使用场景：
2.一、耗时较短的任务。
2.二、任务处理速度 > 任务提交速度 ，这样才能保证不会不断建立新的进程，避免内存被占满。newCachedThreadPool能够当作newFixedThreadPool(无穷大),虽然没法限制线程总数,可是能够减小没必要要的线程建立和销毁上的消耗,

三、取名为cached-threadpool的缘由在于线程池中的线程是被线程池缓存了的，也就是说，线程没有任务要执行时，便处于空闲状态，处于空闲状态的线程并不会被当即销毁（会被缓存住），只有当空闲时间超出一段时间(默认为60s)后，线程池才会销毁该线程（至关于清除过期的缓存）。新任务到达后，线程池首先会让被缓存住的线程（空闲状态）去执行任务，若是没有可用线程（无空闲线程），便会建立新的线程。

 

四、

看一段测试代码：

 

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class newCachedThreadPoolTest {

    public static void main(String[] args) {
    　　ExecutorService executorService = Executors.newCachedThreadPool();
    　　for (int i = 1; i < 10000; i++)
        　　executorService.submit(new task());
    　　}
　　}

　　class task implements Runnable {
    　　@Override
    　　public void run() {
        　　try {
            　　Thread.sleep(5000);
        　　} catch (InterruptedException e) {
            　　e.printStackTrace();
    　　　　}
　　}
}

 

运行结果为：Exception in thread "main" java.lang.OutOfMemoryError: unable to create new native thread

能够看出来是堆外内存溢出，由于咱们新建的线程都在工做（代码中用sleep表示在工做中），newCachedThreadPool 只会重用空闲而且可用的线程，因此上述代码只能不停地建立新线程，在 64-bit JDK 1.7 中 -Xss 默认是 1024k，也就是 1M，那就是须要 10000*1M = 10G 的堆外内存空间来给线程使用，可是个人机器总共就 8G 内存，不够建立新的线程，因此就 OOM 了。

 

因此这个newCachedThreadPool 你们通常不用就是这样的缘由，由于它的最大值是在初始化的时候设置为Integer.MAX_VALUE，通常来讲机器都没那么大内存给它不断使用。固然知道可能出问题的点，就能够去重写一个方法限制一下这个最大值，可是出于后期维护缘由，通常来讲用 newFixedThreadPool 也就足够了。

 

Java如何依据cpu核数设置合适的线程数

newFixedThreadPool 定长线程池的大小最好根据系统资源进行设置。如Runtime.getRuntime().availableProcessors()获取cpu核心数。

executorservice pool = executors.newfixedthreadpool(runtime.getruntime().availableprocessors()*10);通常建议每cpu不超过25个线程，固然多一点也没什么问题，不过就费点线程资源.

java.lang.Runtime.availableProcessors() 方法返回到Java虚拟机的可用的处理器数量。此值可能会改变在一个特定的虚拟机调用。应用程序可用处理器的数量是敏感的，所以偶尔查询该属性，并适当地调整本身的资源使用状况.