JAVA學(xué)習(xí)點(diǎn)點(diǎn)滴滴

synchronized （ lock ） {

? ?? // 執(zhí)行對(duì)共享資源的操作

??? ……

}

public class ServerThread {

????? Object concLock = new Object();

????? int count = 2;

public void runTwoThreads() {

????? // 啟動(dòng)兩個(gè)線程去初始化組件

??????????? new Thread(new ComponentThread1(this)).start();

??????????? new Thread(new ComponentThread1(this)).start();

??????????? // Wait for other thread

while(count != 0) {

????????????????? synchronized(concLock) {

??????????????????????? try {

????????????????????????????? concLock.wait();

????????????????????????????? System.out.println("Wake up.");

??????????????????????? } catch (InterruptedException ie) { // 處理異常 }

????????????????? }

??????????? }

??????????? System.out.println("Server is up.");

????? }

????? public void callBack() {

synchronized(concLock) {

????????????????? count--;

????????????????? concLock.notifyAll();

??????????? }

????? }

????? public static void main(String[] args){

??????????? ServerThread server = new ServerThread();

??????????? server.runTwoThreads();

????? }

}

?

public class ComponentThread1 implements Runnable {

????? private ServerThread server;

????? public ComponentThread1(ServerThread server) {

??????????? this.server = server;

????? }

public void run() {

????? // 做組件初始化的工作

??????????? System.out.println("Do component initialization.");

??????????? server.callBack();

????? }

}

public class DoCallStuff implements Callable<String>{ // *1

??????? private int aInt;

??????? public DoCallStuff(int aInt) {

??????????????? this.aInt = aInt;

??????? }

??????? public String call() throws Exception { //*2

??????????????? boolean resultOk = false;

??????????????? if(aInt == 0){

??????????????????????? resultOk = true;

??????????????? }? else if(aInt == 1){

??????????????????????? while(true){ //infinite loop

?????????????????? ?????????????System.out.println("looping....");

??????????????????????????????? Thread.sleep(3000);

??????????????????????? }

??????????????? } else {

??????????????????????? throw new Exception("Callable terminated with Exception!"); //*3

??????????????? }

??????????????? if(resultOk){

??????????????????????? return "Task done.";

??????????????? } else {

??????????????????????? return "Task failed";

??????????????? }

??????? }

}

import java.util.concurrent.ExecutionException;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

import java.util.concurrent.Future;

public class Executor {

??????? public static void main(String[] args){

??????????????? //*1

??????????????? DoCallStuff call1 = new DoCallStuff(0);

??????????????? DoCallStuff call2 = new DoCallStuff(1);

??????????????? DoCallStuff call3 = new DoCallStuff(2);

??????????????? //*2

??????????????? ExecutorService es = Executors.newFixedThreadPool(3);

??????????????? //*3

??????????????? Future<String> future1 = es.submit(call1);

??????????????? Future<String> future2 = es.submit(call2);

???? ???????????Future<String> future3 = es.submit(call3);

??????????????? try {

??????????????????????? //*4

??????????????????????? System.out.println(future1.get());

???????????????????????? //*5

??????????????????????? Thread.sleep(3000);

????????????????? ??????System.out.println("Thread 2 terminated? :" + future2.cancel(true));

??????????????????????? //*6

??????????????????????? System.out.println(future3.get());

??????????????? } catch (ExecutionException ex) {

??????????????????????? ex.printStackTrace();

??????????????? } catch (InterruptedException ex) {

??????????????????????? ex.printStackTrace();

??????????????? }

??????? }

}

looping....

Task done. //*1

looping....

looping....//*2

looping....

looping....

looping....

looping....

Thread 2 terminated? :true //*3

//*4

java.util.concurrent.ExecutionException: java.lang.Exception: Callable terminated with Exception!

??????? at java.util.concurrent.FutureTask$Sync.innerGet(FutureTask.java:205)

??????? at java.util.concurrent.FutureTask.get(FutureTask.java:80)

??????? at concurrent.Executor.main(Executor.java:43)

??????? …….

public class ScheduledExecutorServiceTest {

??????? public static void main(String[] args)

?????????????? throws InterruptedException, ExecutionException{

?????????????? //*1

??????????????? ScheduledExecutorService service = Executors.newScheduledThreadPool(2);

??????????????? //*2

??????????????? Runnable task1 = new Runnable() {

???????????????????? public void run() {

??????????????????????? System.out.println("Task repeating.");

???????????????????? }

??????????????? };

??????????????? //*3

??????????????? final ScheduledFuture future1 =

??????????????????????? service.scheduleAtFixedRate(task1, 0, 1, TimeUnit.SECONDS);

??????????????? //*4

??????????????? ScheduledFuture<String> future2 = service.schedule(new Callable<String>(){

???????????????????? public String call(){

???????????????????????????? future1.cancel(true);

???????????????????????????? return "task cancelled!";

??????????????? ?????}

??????????????? }, 5, TimeUnit.SECONDS);

??????????????? System.out.println(future2.get());

//*5

service.shutdown();

??????? }

}

//Single Threaded ExecutorService

???? ExecutorService singleThreadeService = Executors.newSingleThreadExecutor();

//Cached ExecutorService

???? ExecutorService cachedService = Executors.newCachedThreadPool();

//Fixed number of ExecutorService

???? ExecutorService fixedService = Executors.newFixedThreadPool(3);

//Single ScheduledExecutorService

???? ScheduledExecutorService singleScheduledService =

????????? Executors.newSingleThreadScheduledExecutor();

//Fixed number of ScheduledExecutorService

ScheduledExecutorService fixedScheduledService =

???? Executors.newScheduledThreadPool(3);

// 生成一個(gè)鎖

Lock lock = new ReentrantLock();

public void accessProtectedResource() {

? lock.lock(); // 取得鎖定

? try {

??? // 對(duì)共享資源進(jìn)行操作

? } finally {

??? // 一定記著把鎖取消掉，鎖本身是不會(huì)自動(dòng)解鎖的

??? lock.unlock() ;

? }

}

public class FileOperator{

????? // 初始化一個(gè) ReadWriteLock

????? ReadWriteLock lock = new ReentrantReadWriteLock();

public String read() {

????? // 得到 readLock 并鎖定

??????????? Lock readLock = lock.readLock();

?????? ?????readLock.lock();

??????????? try {

????????????????? // 做讀的工作

????????????????? return "Read something";

??????????? } finally {

???????????????? readLock.unlock();

??????????? }

????? }

?????

public void write(String content) {

????? // 得到 writeLock 并鎖定

??????????? Lock writeLock = lock.writeLock();

??????????? writeLock.lock();

??????????? try {

????????????????? // 做讀的工作

??????????? } finally {

???????????????? writeLock.unlock();

??????????? }

????? }

}

public class Basket {?????

Lock lock = new ReentrantLock();

// 產(chǎn)生 Condition 對(duì)象

???? Condition produced = lock.newCondition();

???? Condition consumed = lock.newCondition();

???? boolean available = false;

?? ??

???? public void produce() throws InterruptedException {

?????????? lock.lock();

?????????? try {

???????????????? if(available){

??????????????????? consumed.await(); // 放棄 lock 進(jìn)入睡眠 ?

???????????????? }

???????????????? /* 生產(chǎn)蘋(píng)果 */

???????????????? System.out.println("Apple produced.");

???????????????? available = true;

???????????????? produced.signal(); // 發(fā)信號(hào)喚醒等待這個(gè) Condition 的線程

?????????? } finally {

???????????????? lock.unlock();

?????????? }

???? }

????

???? public void consume() throws InterruptedException {

?????????? lock.lock();

?????????? try {

???????????????? if(!available){

?????????????????????? produced.await();// 放棄 lock 進(jìn)入睡眠 ?

???????????????? }

???????????????? /* 吃蘋(píng)果 */

???????????????? System.out.println("Apple consumed.");

???????????????? available = false;

???????????????? consumed.signal();// 發(fā)信號(hào)喚醒等待這個(gè) Condition 的線程

?????????? } finally {

???????????????? lock.unlock();

?????????? }

???? }?????

}

public class ConditionTester {

?????

????? public static void main(String[] args) throws InterruptedException{

final Basket basket = new Basket();

// 定義一個(gè) producer

??????????? Runnable producer = new Runnable() {

????????????????? public void run() {

??????????????????????? try {

????????????????????????????? basket.produce();

?????? ?????????????????} catch (InterruptedException ex) {

????????????????????????????? ex.printStackTrace();

??????????????????????? }

????????????????? }

};

// 定義一個(gè) consumer

??????????? Runnable consumer = new Runnable() {

????????????????? public void run() {

??????????????????????? try {

????????????????????????????? basket.consume();

??????????????????????? } catch (InterruptedException ex) {

????????????????????????????? ex.printStackTrace();

??????????????????????? }

????????????????? }

};

// 各產(chǎn)生 10 個(gè) consumer 和 producer

??????????? ExecutorService service = Executors.newCachedThreadPool();

??????????? for(int i=0; i < 10; i++)

????????????????? service.submit(consumer);

??????????? Thread.sleep(2000);

??????????? for(int i=0; i<10; i++)

????????????????? service.submit(producer);

??????????? service.shutdown();

????? }?????

}

public class Pool {

??? ??ArrayList<String> pool = null;

????? Semaphore pass = null;

????? public Pool(int size){

??????????? // 初始化資源池

??????????? pool = new ArrayList<String>();

??????????? for(int i=0; i<size; i++){

????????????????? pool.add("Resource "+i);

??????????? }

??? ???????? //Semaphore 的大小和資源池的大小一致

??????????? pass = new Semaphore(size);

????? }

????? public String get() throws InterruptedException{

??????????? // 獲取通行證 , 只有得到通行證后才能得到資源

??????????? pass.acquire();

??????????? return getResource();

????? }

????? public void put(String resource){

??????????? // 歸還通行證，并歸還資源

??????????? pass.release();

??????????? releaseResource(resource);

????? }

???? private synchronized String getResource() {

??????????? String result = pool.get(0);

??????????? pool.remove(0);

??????????? System.out.println("Give out "+result);

??????????? return result;

????? }

????? private synchronized void releaseResource(String resource) {

??????????? System.out.println("return "+resource);

??????????? pool.add(resource);

????? }

}

public class SemaphoreTest {

????? public static void main(String[] args){

??????????? final Pool aPool = new Pool(2);

??????????? Runnable worker = new Runnable() {

????????????????? public void run() {

??????????????????????? String resource = null;

??????? ????????????????try {

????????????????????????????? // 取得 resource

????????????????????????????? resource = aPool.get();

??????????????????????? } catch (InterruptedException ex) {

????????????????????????????? ex.printStackTrace();

??????????????????????? }

??????????????????????? // 用 resource 做工作

??????????????????????? System.out.println("I worked on "+resource);

??????????????????????? // 歸還 resource

??????????????????????? aPool.put(resource);

????????????????? }

??????????? };

??????????? ExecutorService service = Executors.newCachedThreadPool();

??????????? for(int i=0; i<20; i++){

????????????????? service.submit(worker);

??????????? }

??????????? service.shutdown();

????? }????

}

public class Server {

????? public static void main(String[] args) throws InterruptedException{

??????????? System.out.println("Server is starting.");

?????????? ?// 初始化一個(gè)初始值為 3 的 CountDownLatch

??????????? CountDownLatch latch = new CountDownLatch(3);

??????????? // 起 3 個(gè)線程分別去啟動(dòng) 3 個(gè)組件

??????????? ExecutorService service = Executors.newCachedThreadPool();

??????????? service.submit(new ComponentThread(latch, 1));

????????? ??service.submit(new ComponentThread(latch, 2));

??????????? service.submit(new ComponentThread(latch, 3));

??????????? service.shutdown();

??????????? // 進(jìn)入等待狀態(tài)

??????????? latch.await();

??????????? // 當(dāng)所需的三個(gè)組件都完成時(shí)， Server 就可繼續(xù)了

??????????? System.out.println("Server is up!");

????? }

}

?

public class ComponentThread implements Runnable{

????? CountDownLatch latch;

????? int ID;

????? /** Creates a new instance of ComponentThread */

????? public ComponentThread(CountDownLatch latch, int ID) {

??????????? this.latch = latch;

??????????? this.ID = ID;

????? }

????? public void run() {

??????????? System.out.println("Component "+ID + " initialized!");

??????????? // 將計(jì)數(shù)減一

??????????? latch.countDown();

????? }????

}

Server is starting.

Component 1 initialized!

Component 3 initialized!

Component 2 initialized!

Server is up!

public class MainThread {

public static void main(String[] args)

????? throws InterruptedException, BrokenBarrierException, TimeoutException{

??????????? final int[] array = new int[2];

??????????? CyclicBarrier barrier = new CyclicBarrier(2,

????????????????? new Runnable() {// 在所有線程都到達(dá) Barrier 時(shí)執(zhí)行

????????????????? public void run() {

??????????????????????? System.out.println("Total is:"+(array[0]+array[1]));

????????????????? }

??????????? });???????????

??????????? // 啟動(dòng)線程

??????????? new Thread(new ComponentThread(barrier, array, 0)).start();

??????????? new Thread(new ComponentThread(barrier, array, 1)).start();???

????? }?????

}

?

public class ComponentThread implements Runnable{

????? CyclicBarrier barrier;

????? int ID;

????? int[] array;

????? public ComponentThread(CyclicBarrier barrier, int[] array, int ID) {

??????????? this.barrier = barrier;

??????????? this.ID = ID;

??????????? this.array = array;

????? }

????? public void run() {

??????????? try {

????????????????? array[ID] = new Random().nextInt();

????????????????? System.out.println(ID+ " generates:"+array[ID]);

????????????????? // 該線程完成了任務(wù)等在 Barrier 處

????????????????? barrier.await();

??????????? } catch (BrokenBarrierException ex) {

????????????????? ex.printStackTrace();

??????????? } catch (InterruptedException ex) {

????????????????? ex.printStackTrace();

??????????? }

????? }

}

class FillAndEmpty {

????? // 初始化一個(gè) Exchanger ，并規(guī)定可交換的信息類(lèi)型是 DataCup

????? Exchanger<Cup> exchanger = new Exchanger();

???? ?Cup initialEmptyCup = ...; // 初始化一個(gè)空的杯子

????? Cup initialFullCup = ...; // 初始化一個(gè)裝滿水的杯子

????? // 服務(wù)生線程

????? class Waiter implements Runnable {

??????????? public void run() {

????????????????? Cup currentCup = initialEmptyCup;

????????????????? try {

?????? ?????????????????// 往空的杯子里加水

??????????????????????? currentCup.addWater();

??????????????????????? // 杯子滿后和顧客的空杯子交換

??????????????????????? currentCup = exchanger.exchange(currentCup);

????????????????? } catch (InterruptedException ex) { ... handle ... }

? ?????????? }

????? }

????? // 顧客線程

????? class Customer implements Runnable {

??????????? public void run() {

????????????????? DataCup currentCup = initialFullCup;

????????????????? try {

??????????????????????? // 把杯子里的水喝掉

??????????????????????? currentCup.drinkFromCup();

??????????????????????? // 將空杯子和服務(wù)生的滿杯子交換

??????????????????????? currentCup = exchanger.exchange(currentCup);

????????????????? } catch (InterruptedException ex) { ... handle ...}

??????????? }

????? }

?????

????? void start() {

??????? ????new Thread(new Waiter()).start();

??????????? new Thread(new Customer()).start();

????? }

}

public class BlockingQueueTest {

????? static BlockingQueue<String> basket;

????? public BlockingQueueTest() {

??????????? // 定義了一個(gè)大小為 2 的 BlockingQueue ，也可根據(jù)需要用其他的具體類(lèi)

??????????? basket = new ArrayBlockingQueue<String>(2);

????? }

????? class Producor implements Runnable {

??????????? public void run() {

????????????????? while(true){

??????????????????????? try {

????????????????????????????? // 放入一個(gè)對(duì)象，若 basket 滿了，等到 basket 有位置

?????????????????????? ???????basket.put("An apple");

??????????????????????? } catch (InterruptedException ex) {

????????????????????????????? ex.printStackTrace();

??????????????????????? }

????????????????? }

??????????? }

????? }

????? class Consumer implements Runnable {

? ??????????public void run() {

????????????????? while(true){

??????????????????????? try {

????????????????????????????? // 取出一個(gè)對(duì)象，若 basket 為空，等到 basket 有東西為止

????????????????????????????? String result = basket.take();

??????????????????????? } catch (InterruptedException ex) {

????????????????????????????? ex.printStackTrace();

??????????????????????? }

????????????????? }

????? ??????}???????????

????? }

????? public void execute(){

??????????? for(int i=0; i<10; i++){

????????????????? new Thread(new Producor()).start();

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

??????????? }???????????

????? }

????? public static void main(String[] args){

??????????? BlockingQueueTest test = new BlockingQueueTest();

??????????? test.execute();

????? }?????

}