java多线程学习--java.util.concurrent

package thread;

import java.util.concurrent.CountDownLatch;

public class TestCountDownLatch {
private static final int N = 10;

public static void main(String[] args) throws InterruptedException {
CountDownLatch doneSignal = new CountDownLatch(N);
CountDownLatch startSignal = new CountDownLatch(1);// 开始执行信号

for (int i = 1; i <= N; i++) {
new Thread(new Worker(i, doneSignal, startSignal)).start();// 线程启动了
}
System.out.println("begin------------");
startSignal.countDown();// 开始执行啦
doneSignal.await();// 等待所有的线程执行完毕
System.out.println("Ok");

}

static class Worker implements Runnable {
private final CountDownLatch doneSignal;
private final CountDownLatch startSignal;
private int beginIndex;

Worker(int beginIndex, CountDownLatch doneSignal,
CountDownLatch startSignal) {
this.startSignal = startSignal;
this.beginIndex = beginIndex;
this.doneSignal = doneSignal;
}

public void run() {
try {
startSignal.await(); // 等待开始执行信号的发布
beginIndex = (beginIndex - 1) * 10 + 1;
for (int i = beginIndex; i < beginIndex + 10; i++) {
System.out.println(i);
}
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
doneSignal.countDown();
}
}
}
}

public class TestCyclicBarrier {
public static void main(String[] args) {
//new 一个线程池
ExecutorService exec = Executors.newCachedThreadPool();
final Random random = new Random();

final CyclicBarrier barrier = new CyclicBarrier(4, new Runnable() {
@Override
public void run() {
System.out.println("大家都到齐了，开始happy去");
}
});

for (int i = 0; i < 4; i++) {
exec.execute(new Runnable() {
@Override
public void run() {
try {
Thread.sleep(random.nextInt(10000));
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(Thread.currentThread().getName()
+ "到了，其他哥们呢");
try {
barrier.await();// 等待其他哥们
} catch (InterruptedException e) {
e.printStackTrace();
} catch (BrokenBarrierException e) {
e.printStackTrace();
}
}
});
}
exec.shutdown();
}
}

acquire()

release()

Semaphore

public class TestSemaphore {
public static void main(String[] args) {
ExecutorService exec = Executors.newCachedThreadPool();
TestSemaphore t = new TestSemaphore();
final BoundedHashSet<String> set = t.getSet();

for (int i = 0; i < 3; i++) {// 三个线程同时操作add
exec.execute(new Runnable() {
public void run() {
try {
set.add(Thread.currentThread().getName());
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
}

for (int j = 0; j < 3; j++) {// 三个线程同时操作remove
exec.execute(new Runnable() {
public void run() {
set.remove(Thread.currentThread().getName());
}
});
}
exec.shutdown();
}

public BoundedHashSet<String> getSet() {
return new BoundedHashSet<String>(2);// 定义一个边界约束为2的线程
}

class BoundedHashSet<T> {
private final Set<T> set;
private final Semaphore semaphore;

public BoundedHashSet(int bound) {
this.set = Collections.synchronizedSet(new HashSet<T>());//①
this.semaphore = new Semaphore(bound, true);
}

public void add(T o) throws InterruptedException {
semaphore.acquire();// 信号量控制可访问的线程数目
set.add(o);
System.out.printf("add:%s%n", o);
}

public void remove(T o) {
if (set.remove(o))
semaphore.release();// 释放掉信号量
System.out.printf("remove:%s%n", o);
}
}
}

Collection c = Collections.synchronizedCollection(new ArrayList());
List l = Collections.synchronizedList(new ArrayList());
Set s = Collections.synchronizedSet(new HashSet());
Map m = Collections.synchronizedMap(new HashMap());

Future

get

runAndReset()

public class TestFutureTask {
public static void main(String[] args) {
ExecutorService exec = Executors.newCachedThreadPool();

FutureTask<String> task = new FutureTask<String>(
new Callable<String>() {// FutrueTask的构造参数是一个Callable接口
@Override
public String call() throws Exception {
return Thread.currentThread().getName();// 这里可以是一个异步操作
}
});

try {
exec.execute(task);// FutureTask实际上也是一个线程
String result = task.get();// 取得异步计算的结果，如果没有返回，就会一直阻塞等待
System.out.printf("get:%s%n", result);
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
}
}

package thread;

import java.util.ArrayList;
import java.util.concurrent.Exchanger;
/**
*    1. Exchanger用于在2个线程中交换对象。
2. return_object = exchanger.exchange(exch_object)
3. 例子中Producer向ArrayList中缓慢填充随机整数，Consumer从另一个ArrayList中缓慢取出整数并输出。
4. 当Producer的ArrayList填满，并且Consumer的ArrayList为空时，2个线程才交换ArrayList。
* @author Administrator
*/
public class ExchangerTest {

private static Exchanger<ArrayList<Integer>> exchanger = null;
private static ArrayList<Integer> buffer1 = null;
private static ArrayList<Integer> buffer2 = null;

public static void main(String[] args) throws Exception {
exchanger = new Exchanger<ArrayList<Integer>>();
buffer1 = new ArrayList<Integer>(10);
buffer2 = new ArrayList<Integer>(10);

Thread pth = new ProducerThread();
Thread cth = new ConsumerThread();

pth.start();
cth.start();

Thread.sleep(60 * 1000);
System.out.println("main: interrupting threads.");
pth.interrupt();
cth.interrupt();

pth.join();
cth.join();

System.out.println("main: end.");
}

private static class ProducerThread extends Thread {
@Override
public void run() {
ArrayList<Integer> buff = buffer1;
try {
while (true) {
if (buff.size() >= 10) {
// 与consumer交换buffer.
System.out.println("producer: exchanging.");
buff = exchanger.exchange(buff);
buff.clear();
}

// 随机产生一个0-100的整数。
int x = (int) (Math.random() * 100);
buff.add(x);
System.out.println("producer: " + x);

// 随机等待0-3秒 。
int t = (int) (Math.random() * 3);
Thread.sleep(t * 1000);
}
} catch (InterruptedException e) {
System.out.println("producer: interrupted.");
}
}
}

private static class ConsumerThread extends Thread {
@Override
public void run() {
ArrayList<Integer> buff = buffer2;
try {
while (true) {
for (Integer x : buff) {
System.out.println("consumer: " + x);

// 随机等待0-3秒 。
int t = (int) (Math.random() * 3);
Thread.sleep(t * 1000);
}

// 与producer交换buffer。
System.out.println("consumer: exchanging.");
buff = exchanger.exchange(buff);
}
} catch (InterruptedException e) {
System.out.println("consumer: interrupted.");
}
}
}
}

import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class TestScheduledThreadPoolExecutor {

public static void main(String[] args) {
ScheduledThreadPoolExecutor exec = new ScheduledThreadPoolExecutor(1);
exec.scheduleAtFixedRate(new Runnable() {// 每隔一段时间就触发异常
@Override
public void run() {
throw new RuntimeException();
}
}, 1000, 5000, TimeUnit.MILLISECONDS);
exec.scheduleAtFixedRate(new Runnable() {// 每隔一段时间打印系统时间，证明两者是互不影响的
@Override
public void run() {
System.out.println(System.nanoTime());
}
}, 1000, 2000, TimeUnit.MILLISECONDS);
}

}

Queue

package thread;

import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

public class BlockingQueueTest {

public static void main(String[] args) {
final BlockingQueue<Integer> queue = new LinkedBlockingQueue<Integer>(3);
final Random random = new Random();

class Producer implements Runnable {
@Override
public void run() {
while (true) {
try {
int i = random.nextInt(100);
queue.put(i);// 当队列达到容量时候，会自动阻塞的
if (queue.size() == 3) {
System.out.println("full");
Thread.sleep(1000);
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}

class Consumer implements Runnable {
@Override
public void run() {
while (true) {
try {
queue.take();// 当队列为空时，也会自动阻塞
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}

new Thread(new Producer()).start();
new Thread(new Consumer()).start();

}

}

package thread;
import java.util.Random;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;
public class TestDelayQueue {
private class Stadium implements Delayed {
long trigger;
public Stadium(long i) {
trigger = System.currentTimeMillis() + i;
}
@Override
public long getDelay(TimeUnit arg0) {
long n = trigger - System.currentTimeMillis();
return n;
}
@Override
public int compareTo(Delayed arg0) {
return (int) (this.getDelay(TimeUnit.MILLISECONDS) - arg0
.getDelay(TimeUnit.MILLISECONDS));
}
public long getTriggerTime() {
return trigger;
}
}
public static void main(String[] args) throws Exception {
Random random = new Random();
DelayQueue<Stadium> queue = new DelayQueue<Stadium>();
TestDelayQueue t = new TestDelayQueue();
for (int i = 0; i < 5; i++) {
queue.add(t.new Stadium(random.nextInt(30000)));
}
Thread.sleep(2000);
while (true) {
Stadium s = queue.take();// 延时时间未到就一直等待
if (s != null) {
System.out.println(System.currentTimeMillis()
- s.getTriggerTime());// 基本上是等于0
}
if (queue.size() == 0)
break;
}
}
}

package thread;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.SynchronousQueue;
public class TestSynchronousQueue {
class Producer implements Runnable {
private BlockingQueue<String> queue;
List<String> objects = Arrays.asList("one", "two", "three");
public Producer(BlockingQueue<String> q) {
this.queue = q;
}
@Override
public void run() {
try {
for (String s : objects) {
queue.put(s);// 产生数据放入队列中
System.out.printf("put:%s%n", s);
}
queue.put("Done");// 已完成的标志
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
class Consumer implements Runnable {
private BlockingQueue<String> queue;
public Consumer(BlockingQueue<String> q) {
this.queue = q;
}
@Override
public void run() {
String obj = null;
try {
while (!((obj = queue.take()).equals("Done"))) {
System.out.println(obj);// 从队列中读取对象
Thread.sleep(3000); // 故意sleep，证明Producer是put不进去的
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
public static void main(String[] args) {
BlockingQueue<String> q = new SynchronousQueue<String>();
TestSynchronousQueue t = new TestSynchronousQueue();
new Thread(t.new Producer(q)).start();
new Thread(t.new Consumer(q)).start();
}
}