public synchronized ExecutorService executorService() {
if (executorService == null) {
executorService = new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>(), Util.threadFactory("OkHttp Dispatcher", false));
}
return executorService;
}
ThreadPoolExecutor的参数
- corePoolSize :0 核心并发数，就是在线程池不饱和时，线程池可拥有的线程数。如果是0的话，空闲一段时间后所有线程将全部被销毁。
-maximumPoolSize：线程池最大线程容量。
-keepAliveTIme: 当总线程数大于核心线程数corePoolSize那部分线程存活的时间。
-BlockingQueue<Runnable>:
这个参数被称为阻塞队列(生产者消费者模型)
1.ArrayBlockingQueue
2.LinkedBlockingQueue
上此两个要注意指定最大容量，如果生产者的效率很高，会把队列缓存占满，然而没有指定最大值会消耗掉内存
3.PriorityBlockingQueue
4.DelayQueue
5.SynchronousQueue
它是一个不存储元素的阻塞队列。每个插入操作必须等待另一个线程的移除操作，同样移除操作也是如此。因此队列中没有存储一个元素。（多线程打印出0101010101）
练习下队列，看有毛用
public class BlockQueueTest {
private static final int QUEUESIZE= 1;
private ArrayBlockingQueue<Integer> integers;
@Test
public void BlockQueueTest(){
integers = new ArrayBlockingQueue<>(QUEUESIZE);
Consumer consumer = new Consumer();
Producer producer = new Producer();
consumer.start();
producer.start();
}
class Consumer extends Thread{
@Override
public void run() {
super.run();
while (true)
{
try {
Integer take = integers.take();
System.out.println("消费元素:"+take);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
class Producer extends Thread{
@Override
public void run() {
super.run();
while (true)
{
try {
integers.put(1);
System.out.println("生产元素int 1");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
}
打印010101 顺序一个没乱
package com.system.bhouse.bhouse.Queue;
import org.junit.Test;
import java.util.concurrent.SynchronousQueue;
/**
* Created by wz on 2018-11-11.
*/
public class SynchronousQueueTest {
private static final int QUEUESIZE= 1;
private SynchronousQueue<Integer> integers;
private volatile boolean isConsumer = false;
@Test
public void BlockQueueTest(){
integers = new SynchronousQueue<>();
Consumer consumer = new Consumer();
Producer producer = new Producer();
consumer.start();
producer.start();
}
class Consumer extends Thread{
@Override
public void run() {
super.run();
synchronized (SynchronousQueueTest.this) {
while (true)
{
if (isConsumer) {
System.out.println("消费元素:" + 1);
isConsumer = !isConsumer;
SynchronousQueueTest.this.notify();
}else {
try {
SynchronousQueueTest.this.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
}
}
class Producer extends Thread{
@Override
public void run() {
super.run();
synchronized (SynchronousQueueTest.this) {
while (true)
{
if (!isConsumer) {
System.out.println("生产元素int 0");
isConsumer = !isConsumer;
SynchronousQueueTest.this.notify();
}else {
try {
SynchronousQueueTest.this.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
}
}
}
以为SynchronousQueue可以一对一通信配置，应该是打印01010101的最佳配置，发现你无法控制队列里的同步机制。代码运行到那队列操作就停止了。
package com.system.bhouse.bhouse.Queue;
import org.junit.Test;
import java.util.concurrent.SynchronousQueue;
/**
* Created by wz on 2018-11-11.
*/
public class SynchronousQueueTest2 {
private SynchronousQueue<Integer> integers;
private volatile boolean isConsumer = false;
@Test
public void BlockQueueTest(){
integers = new SynchronousQueue<>();
Consumer consumer = new Consumer();
Producer producer = new Producer();
consumer.start();
producer.start();
}
class Consumer extends Thread{
@Override
public void run() {
super.run();
while (true)
{
if (isConsumer) {
try {
Integer take = integers.take();
System.out.println(take);
isConsumer=!isConsumer;
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
}
class Producer extends Thread {
@Override
public void run() {
super.run();
while (true) {
if (!isConsumer) {
try {
isConsumer = !isConsumer;
integers.put(1);
System.out.println(0);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
}
}
言归正传
public synchronized ExecutorService executorService() {
if (executorService == null) {
executorService = new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>(), Util.threadFactory("OkHttp Dispatcher", false));
}
return executorService;
}
来测试一下这个线程池特点
public class CacheTheadPool {
private static ThreadPoolExecutor executorService;
@Test
public void cacheTheadPool(){
// for (int i=0;i<100;i++) {
// newCachedThreadPool().execute(new AsyncCall("thread"+i));
// }
newCachedThreadPool().execute(new AsyncCall("thread"+1));
newCachedThreadPool().execute(new AsyncCall("thread"+2));
newCachedThreadPool().execute(new AsyncCall("thread"+3));
System.out.println("先开3个，按书上讲会有3个是新建线程");
System.out.println("线程池核心："+executorService.getCorePoolSize());
System.out.println("线程池数目:"+executorService.getPoolSize());
System.out.println("队列任务数目:"+executorService.getQueue().size());
//让上面的用完
// try {
// Thread.sleep(500);
// } catch (InterruptedException e) {
// e.printStackTrace();
// }
newCachedThreadPool().execute(new AsyncCall("thread"+4));
newCachedThreadPool().execute(new AsyncCall("thread"+5));
newCachedThreadPool().execute(new AsyncCall("thread"+6));
System.out.println("再开3个，按书上讲会有3个是,看看是不是复用");
System.out.println("线程池核心："+executorService.getCorePoolSize());
System.out.println("线程池数目:"+executorService.getPoolSize());
System.out.println("队列任务数目:"+executorService.getQueue().size());
try {
Thread.sleep(8000);
} catch (InterruptedException e) {
e.printStackTrace();
}
newCachedThreadPool().execute(new AsyncCallSleep("thread"+7));
newCachedThreadPool().execute(new AsyncCallSleep("thread"+8));
newCachedThreadPool().execute(new AsyncCallSleep("thread"+9));
System.out.println("再开3个，按书上讲会有3个是,看看是不是新建");
System.out.println("线程池核心："+executorService.getCorePoolSize());
System.out.println("线程池数目:"+executorService.getPoolSize());
System.out.println("队列任务数目:"+executorService.getQueue().size());
}
/**
* 建立的都是 用户线程 优先级比较高.
* @return
*/
public synchronized ExecutorService newCachedThreadPool(){
if (executorService == null) {
executorService = new ThreadPoolExecutor(0, 6, 5, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>(), Util.threadFactory("OkHttp Dispatcher", false));
}
return executorService;
}
final class AsyncCall extends NamedRunnable {
private AsyncCall(Object... arg){
super("OkHttp %s",arg);
}
@Override
protected void execute() {
String name = Thread.currentThread().getName();
System.out.println("当前处理的线程名是:"+name);
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
final class AsyncCallSleep extends NamedRunnable {
private AsyncCallSleep(Object... arg){
super("OkHttpSleep %s",arg);
}
@Override
protected void execute() {
String name = Thread.currentThread().getName();
System.out.println("当前处理的随眠线程名是:"+name);
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
public abstract class NamedRunnable implements Runnable {
protected final String name;
public NamedRunnable(String format, Object... args) {
this.name = String.format(format, args);
}
@Override public final void run() {
String oldName = Thread.currentThread().getName();
Thread.currentThread().setName(name);
try {
execute();
} finally {
Thread.currentThread().setName(oldName);
}
}
protected abstract void execute();
}
}
