- CountDownLatch使用
- CyclicBarrier使用
- Exchanger使用
- Semaphore信号量
- Condition交替打印123，共10次
- 线程main、线程0、线程1按照顺序依次执行
- 线程池内部线程执行任务报错处理
- execute提交的任务(无返回值)
- submit提交的任务报错
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class CountDownLatchDemo {
// 创建2个
static CountDownLatch countDownLatch = new CountDownLatch(2);
// 线程池2个线程
static ExecutorService pool = Executors.newFixedThreadPool(2);
public static void main(String[] args) {
pool.submit(()->{
System.out.println("a thread ... ");
try {
Thread.sleep(2000);
System.out.println("a thread ... end");
}catch (InterruptedException e){
}
// 执行结束调用countDown
countDownLatch.countDown();
});
pool.submit(()->{
System.out.println("b thread ... ");
try {
Thread.sleep(2000);
System.out.println("b thread ...end ");
}catch (InterruptedException e){
}
countDownLatch.countDown();
});
try {
// 主线程在等待a b线程执行结束
countDownLatch.await();
}catch (InterruptedException e){
}
System.out.println("main ...");
// 关闭线程池
pool.shutdown();
}
}
执行结果打印
a thread ... b thread ... b thread ...end a thread ... end main ...CyclicBarrier使用
package wang.thread;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class CyclicBarrierDemo {
static ExecutorService pool = Executors.newFixedThreadPool(2);
static CyclicBarrier cyclicBarrier = new CyclicBarrier(2);
public static void main(String[] args) {
pool.submit(()->{
System.out.println("a exe ...");
//Thread.setDefaultUncaughtExceptionHandler();
try {
Thread.sleep(4000);
cyclicBarrier.await();
System.out.println("a exe ...end");
}catch (InterruptedException e){
}catch (BrokenBarrierException e){
}
});
pool.submit(()->{
System.out.println("b exe ...");
try {
Thread.sleep(4000);
cyclicBarrier.await();
System.out.println("b exe ...end");
}catch (InterruptedException e){
}catch (BrokenBarrierException e){
}
});
System.out.println("main ...");
pool.shutdown();
}
}
Exchanger使用
package wang.thread;
import java.util.concurrent.Exchanger;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class ExchangeDemo {
static ExecutorService pool = Executors.newFixedThreadPool(2);
public static void main(String[] args) {
Exchanger ex = new Exchanger<>();
pool.submit(() -> {
try {
String css = ex.exchange("ctl");
System.out.println("a "+css);
}catch (InterruptedException e){
}
});
pool.submit(() -> {
try {
String ctl = ex.exchange("css");
System.out.println("b "+ctl);
}catch (InterruptedException e){
}
});
pool.shutdown();
}
}
交换结果
b ctl a cssSemaphore信号量
package wang.thread;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
public class SemaphoreDemo {
static Semaphore semaphore = new Semaphore(3);
static ExecutorService pool = Executors.newFixedThreadPool(10);
public static void main(String[] args) {
for(int i=1;i<=10;i++){
final int tmp = i;
pool.submit(()->{
try {
semaphore.acquire();
System.out.println(Thread.currentThread().getName()+" "+tmp+" :exe...");
Thread.sleep(5000);
semaphore.release();
} catch (InterruptedException e) {
e.printStackTrace();
}
},"thread"+tmp+"name");
}
while (semaphore.hasQueuedThreads()){
// 1s 采样一次
System.out.println("queuelength:"+semaphore.getQueueLength());
try {
Thread.sleep(1000);
}catch (InterruptedException e){
}
}
pool.shutdown();
}
}
打印结果，可以看到，虽然有10个任务，但是最多同时只有3个线程执行任务
pool-1-thread-3 3 :exe... pool-1-thread-2 2 :exe... pool-1-thread-1 1 :exe... queuelength:5 queuelength:7 queuelength:7 queuelength:7 queuelength:7 pool-1-thread-4 4 :exe... pool-1-thread-5 5 :exe... pool-1-thread-6 6 :exe... queuelength:4 queuelength:4 queuelength:4 queuelength:4 queuelength:4 pool-1-thread-7 7 :exe... pool-1-thread-8 8 :exe... pool-1-thread-9 9 :exe... queuelength:1 queuelength:1 queuelength:1 queuelength:1 queuelength:1 pool-1-thread-10 10 :exe...Condition交替打印123，共10次
package wang.thread;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
public class ConditionaDemo {
static ReentrantLock lock = new ReentrantLock();
static Condition condition1 = lock.newCondition();
static Condition condition2 = lock.newCondition();
static Condition condition3 = lock.newCondition();
static volatile int flag = 1;
public static void main(String[] args) throws InterruptedException {
Thread thread1 = new Thread(new Print1(), "thread-1");
Thread thread2 = new Thread(new Print2(), "thread-2");
Thread thread3 = new Thread(new Print3(), "thread-3");
thread1.start();
thread2.start();
thread3.start();
thread1.join();
thread2.join();
thread3.join();
System.out.println("main ...");
}
static void print1() {
try {
lock.lock();
while (flag != 1) {
condition1.await();
}
System.out.println(Thread.currentThread().getName() + "print 1");
flag = 2;
condition2.signal();
} catch (Exception e) {
} finally {
lock.unlock();
}
}
static void print2() {
try {
lock.lock();
while (flag != 2) {
condition2.await();
}
System.out.println(Thread.currentThread().getName() + "print 2");
flag = 3;
condition3.signal();
} catch (Exception e) {
} finally {
lock.unlock();
}
}
static void print3() {
try {
lock.lock();
while (flag != 3) {
condition3.await();
}
System.out.println(Thread.currentThread().getName() + "print 3");
flag = 1;
condition1.signal();
} catch (Exception e) {
} finally {
lock.unlock();
}
}
static class Print1 implements Runnable {
@Override
public void run() {
for (int i = 1; i <= 10; i++) {
print1();
}
}
}
static class Print2 implements Runnable {
@Override
public void run() {
for (int i = 1; i <= 10; i++) {
print2();
}
}
}
static class Print3 implements Runnable {
@Override
public void run() {
for (int i = 1; i <= 10; i++) {
print3();
}
}
}
}
线程main、线程0、线程1按照顺序依次执行
package wang.thread;
public class ThreadJoinDemo {
public static void main(String[] args) {
Thread t1 = new Thread(new MyRunable(Thread.currentThread()));
Thread t2 = new Thread(new MyRunable(t1));
t1.start();
t2.start();
try {
System.out.println("main sleep ...");
Thread.sleep(5000);
System.out.println("main sleep ...end");
}catch (InterruptedException e){
}
}
static class MyRunable implements Runnable {
Thread father;
public MyRunable(Thread father) {
this.father = father;
}
@Override
public void run() {
try {
father.join();
} catch (InterruptedException e) {
}
Thread thread = Thread.currentThread();
System.out.println(thread.getName() + "exe...");
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
}
System.out.println(thread.getName() + "exe...end");
}
}
}
打印结果
main sleep ... main sleep ...end Thread-0exe... Thread-0exe...end Thread-1exe... Thread-1exe...end线程池内部线程执行任务报错处理 execute提交的任务(无返回值)
package wang.thread;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;
public class ThreadErrorDemo {
static ExecutorService pool = new ThreadPoolExecutor(4,
4,
1000,
TimeUnit.MILLISECONDS,
new LinkedBlockingDeque<>(),
new BizOneFactory("biz-1"),
new ThreadPoolExecutor.CallerRunsPolicy());
public static void main(String[] args) throws ExecutionException, InterruptedException {
pool.execute(()->{
System.out.println("start ...");
int c = 1/1;
System.out.println("1/1");
int b = 1/0;
});
Future future = pool.submit(()->{
System.out.println("start ...");
int c = 1/1;
System.out.println("1/1");
int b = 1/0;
});
try {
Object o = future.get();
System.out.println(o);
}catch (Exception e){
System.out.println(e.getMessage());
}
pool.shutdown();
}
static class BizOneFactory implements ThreadFactory {
private final String namePrefix;
private final AtomicLong sequenceNo = new AtomicLong(0);
public BizOneFactory(String namePrefix) {
this.namePrefix = namePrefix;
}
@Override
public Thread newThread(Runnable runnable) {
String threadName = String.format("%s-%d", namePrefix, sequenceNo.getAndIncrement());
Thread thread = new Thread(runnable, threadName);
thread.setDaemon(false);
thread.setUncaughtExceptionHandler(new BizOneExceptionHandler());
return thread;
}
}
static class BizOneExceptionHandler implements Thread.UncaughtExceptionHandler {
@Override
public void uncaughtException(Thread t, Throwable e) {
System.out.println("Exception: " + e.getMessage() + " and thread is:" + t.getName());
}
}
}
在线程池创建线程的时候，设置创建线程的factory创建的线程设置setUncaughtExceptionHandler的参数，上述代码自定义了BizOneExceptionHandler类处理异常
submit提交的任务报错// 提交任务
Future future = pool.submit(()->{
System.out.println("start ...");
int c = 1/1;
System.out.println("1/1");
int b = 1/0;
});
try {
// get可能会阻塞,返回的o有可能为null,也有可能get报异常错误
Object o = future.get();
System.out.println(o);
}catch (Exception e){
System.out.println(e.getMessage());
}
像submit提交的任务，get结果的时候要注意。submit提交的任务，看不到具体异常的线程名称
