Java 并发编程实践基础 读书笔记: 第三章 使用 JDK 并发包构建程序

一,JDK并发包实际上就是指java.util.concurrent包里面的那些类和接口等

  主要分为以下几类: 1,原子量;2,并发集合;3,同步器;4,可重入锁;5,线程池

二,原子量

  原子变量主要有AtomicInteger,AtomicLong,AtomicBoolean等,

  主要实现原理都是底层实现类CAS 即比较并交换,都有get,set,compareAndSet等方法,如++,--等也都是有自带方法实现

  这些都是线程安全的,保证了多线程访问时候的可见性

  

 import java.util.concurrent.atomic.AtomicLong;

 /**
* StudySjms
* <p>
* Created by haozb on 2018/3/2.
*/
public class AtomicAccount { AtomicLong account; public AtomicAccount(long money) {
this.account = new AtomicLong(money);
} public void withDraw(long money,int sleepTime){
long oldValue = account.get();
if(oldValue >= money){
try {
Thread.sleep(sleepTime);
} catch (Exception e) { }
if(account.compareAndSet(oldValue,oldValue-money)){
System.out.println(Thread.currentThread().getName()+"扣钱成功");
}else{
System.out.println(Thread.currentThread().getName()+"扣钱失败");
}
}else{
System.out.println("钱不够了");
}
} public static void main(String[] args) {
final AtomicAccount aa = new AtomicAccount(); for (int i = ; i < ; i++) {
new Thread(new Runnable() {
@Override
public void run() {
aa.withDraw(,);
}
}).start();
}
}
}

上面这个方法就是利用原子量解决多线程中计数不安全的例子;

三,并发集合

这个包里面有一个阻塞队列的接口BlockingQueue,阻塞的概念就是满了就不会再填充了,空了也不允许再取,

所有实现这个接口的队列都是线程安全的。

主要有ArrayBlockingQueue:一个由数组支持的有界队列

   LinkedBlockingQueue:一个由链接节点支持的可选有界队列

     PriorityBlockingQueue:一个由优先级堆支持的*优先级队列

   DelayQueue :一个由优先级堆支持的、基于时间的调度队列

ConcurrentMap 接口,ConcurrentHashMap, 这个实现类的方法是原子的,源代码里面是采用lock住put那一块代码。

CopyOnWriteArrayList,CopyOnWriteArraySet采用copy-on-write 模式

package copyonwrite;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.ConcurrentModificationException;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList; public class CopyOnWriteDemo {
@SuppressWarnings("unchecked")
public static void main(String args[]) {
String[] ss = {"aa", "bb", "cc"};
List list1 = new CopyOnWriteArrayList(Arrays.asList(ss));
List list2 = new ArrayList(Arrays.asList(ss));
Iterator itor1 = list1.iterator();
Iterator itor2 = list2.iterator();
list1.add("New");
list2.add("New");
try {
printAll(itor1);
} catch (ConcurrentModificationException e) {
System.err.println("Shouldn't get here");
}
try {
printAll(itor2);
} catch (ConcurrentModificationException e) {
System.err.println("Will gethere.ConcurrentModificationException occurs !");
}
} @SuppressWarnings("unchecked")
private static void printAll(Iterator itor) {
while (itor.hasNext()) {
System.out.println(itor.next());
}
}
}

运行结果如下:
Will get here.ConcurrentModificationException occurs!
aa
bb
cc

这个例子很好地说明了。

四,同步器

  主要有CyclicBarrier:

  1.它允许在涉及一组固定大小的线程的程序中,这些线程必须不时地互相等待

  2.重要的属性就是参与者个数,另外最要方法是 await()。当所有线程都调用了 await()后,就表示这些线程都可以继续执行,否则就会等待

 package synchronizer;

 import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors; public class CyclicBarrierDemo {
/* 徒步需要的时间: Shenzhen, Guangzhou, Chongqing */
private static int[] timeForWalk = { , , };
/* 自驾游 */
private static int[] timeForSelf = { , , };
/* 旅游大巴 */
private static int[] timeForBus = { , , }; static String nowTime() /* 时间格式化 */
{
SimpleDateFormat sdf = new SimpleDateFormat( "HH:mm:ss" );
return(sdf.format( new Date() ) + ": ");
} static class Tour implements Runnable {
private int[] timeForUse;
private CyclicBarrier barrier;
private String tourName; public Tour( CyclicBarrier barrier, String tourName, int[] timeForUse )
{
this.timeForUse = timeForUse;
this.tourName = tourName;
this.barrier = barrier;
} public void run()
{
try {
Thread.sleep( timeForUse[] * );
System.out.println( nowTime() + tourName + " ReachedShenenzh " );
barrier.await(); /* 到达中转站后等待其他旅行团 */
Thread.sleep( timeForUse[] * );
System.out.println( nowTime() + tourName + " ReachedGuangzhou" );
barrier.await(); /* 到达中转站后等待其他旅行团 */
Thread.sleep( timeForUse[] * );
System.out.println( nowTime() + tourName + " ReachedChonin" );
barrier.await(); /* 到达中转站后等待其他旅行团 */
} catch ( InterruptedException e ) {
} catch ( BrokenBarrierException e ) {
}
}
} public static void main( String[] args )
{
/* 三个旅行团都到到达某一个站点后,执行下面的操作,表示都到齐了。 */
Runnable runner = new Runnable()
{
@Override
public void run()
{
System.out.println( "we all are here." );
}
};
CyclicBarrier barrier = new CyclicBarrier( , runner );
/* 使用线程池 */
ExecutorService exec = Executors.newFixedThreadPool( );
exec.submit( new Tour( barrier, "WalkTour", timeForWalk ) );
exec.submit( new Tour( barrier, "SelfTour", timeForSelf ) );
exec.submit( new Tour( barrier, "BusTour", timeForBus ) );
exec.shutdown();
}
}
 ::: SelfTour Reached Shenzhen
: Bus : : Tour Reached Shenzhen
: WalkTour Reached Shenzhen : :
we all are here.
: SelfTour Reached Guangzhou : :
: BusTour Reached Guangzhou : :
: WalkTour Reached Guangzhou : :
we all are here.
: SelfTour Reached Ch : : ongqing
::: BusTour Reached Chongqing
::: WalkTour Reached Chongqing
we all are here.

五,Future(接口)和FutureTask(实现类)

  注:FutureTask实现了Runnable,所以可以通过线程池和Thread执行

  使用他们的好处是可以获得线程的结果,和抛出异常,这种好处通过Callable的定义就知道了

 public interface Callable<V> {
/**
* Computes a result, or throws an exception if unable to do so.
*
* @return computed result
* @throws Exception if unable to compute a result
*/
V call() throws Exception;
}

可以通过以下几种方式调用

 import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask; public class CallableTest { public static void main(String[] args) {
// //创建线程池
// ExecutorService es = Executors.newSingleThreadExecutor();
// //创建Callable对象任务
// CallableDemo calTask=new CallableDemo();
// //提交任务并获取执行结果
// Future<Integer> future =es.submit(calTask);
// //关闭线程池
// es.shutdown(); //创建线程池
ExecutorService es = Executors.newSingleThreadExecutor();
//创建Callable对象任务
CallableDemo calTask=new CallableDemo();
//创建FutureTask
FutureTask<Integer> futureTask=new FutureTask<Integer>(calTask);
//执行任务
es.submit(futureTask);
//关闭线程池
es.shutdown();
try {
Thread.sleep();
System.out.println("主线程在执行其他任务"); if(futureTask.get()!=null){
//输出获取到的结果
System.out.println("futureTask.get()-->"+futureTask.get());
}else{
//输出获取到的结果
System.out.println("futureTask.get()未获取到结果");
} } catch (Exception e) {
e.printStackTrace();
}
System.out.println("主线程在执行完成");
}
}

六,ReentrantLock显示锁 也叫可重入锁; 必须在finally里面释放锁

  实现方式:

 Lock lock=new ReentrantLock();
lock.lock();
try{
// 更新对象状态
}
finally{
lock.unlock();
}

new的时候,有个参数,true或者false 决定了是不是公平锁,正常上不公平锁的性能比较好!

线程之间的交互,有一个类叫Condition:Condition 的方法与 wait 、notify 和 notifyAll 方法类似,分别命名为 await 、 signal 和 signalAll

后面会详细的介绍

七,ReadWriteLock

ReadWriteLock 维护了一对相关的锁,一个用于只读操作,另一个用于写入操作。只要没有 writer,读取锁可以由多个 reader 线程同时保持。写入锁是独占的

这个锁比互斥锁的性能上应该好些(读的频率大于写的频率)

 import java.util.Calendar;
import java.util.Map;
import java.util.TreeMap;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantReadWriteLock; /**
* StudySjms
* <p>
* Created by haozb on 2018/3/2.
*/
public class ReadWriteLockDemo {
private ReentrantReadWriteLock lock = null;
private Lock readLock = null;// 读锁
private Lock writeLock = null;// 写锁
public int key = ;
public int index = ;
public Map<Integer, String> dataMap = null;// 线程共享数据 public ReadWriteLockDemo() {
lock = new ReentrantReadWriteLock(true);
readLock = lock.readLock();
writeLock = lock.writeLock();
dataMap = new TreeMap<Integer, String>();
} public static void main(String[] args) {
ReadWriteLockDemo tester = new ReadWriteLockDemo();
// 第一次获取锁
tester.writeLock.lock();
System.out
.println(Thread.currentThread().getName() + " get writeLock.");
// 第二次获取锁,应为是可重入锁
tester.writeLock.lock();
System.out
.println(Thread.currentThread().getName() + " get writeLock.");
tester.readLock.lock();
System.out.println(Thread.currentThread().getName() + " get readLock");
tester.readLock.lock();
System.out.println(Thread.currentThread().getName() + " get readLock");
tester.readLock.unlock();
tester.readLock.unlock();
tester.writeLock.unlock();
tester.writeLock.unlock();
tester.test();
} public void test() {
// 读线程比写线程多
for (int i = ; i < ; i++) {
new Thread(new reader(this)).start();
}
for (int i = ; i < ; i++) {
new Thread(new writer(this)).start();
}
} public void read() {
// 获取锁
readLock.lock();
try {
if (dataMap.isEmpty()) {
Calendar now = Calendar.getInstance();
System.out.println(now.getTime().getTime() + " R "
+ Thread.currentThread().getName()
+ " get key, but map is empty.");
}
String value = dataMap.get(index);
Calendar now = Calendar.getInstance();
System.out.println(now.getTime().getTime() + " R "
+ Thread.currentThread().getName() + " key = " + index
+ " value = " + value + " map size = " + dataMap.size());
if (value != null) {
index++;
}
} finally {
// 释放锁
readLock.unlock();
}
try {
Thread.sleep();
} catch (Exception e) { }
} public void write() {
writeLock.lock();
try {
String value = "value" + key;
dataMap.put(new Integer(key), value);
Calendar now = Calendar.getInstance();
System.out.println(now.getTime().getTime() + " W "
+ Thread.currentThread().getName() + " key = " + key
+ " value = " + value + " map size = " + dataMap.size());
key++;
try {
Thread.sleep();
} catch (InterruptedException e) {
e.printStackTrace();
}
} finally {
writeLock.unlock();
}
}
} class reader implements Runnable {
private ReadWriteLockDemo tester = null; public reader(ReadWriteLockDemo tester) {
this.tester = tester;
} @Override
public void run() {
Calendar now = Calendar.getInstance();
System.out.println(now.getTime().getTime() + " R "
+ Thread.currentThread().getName() + " started");
for (int i = ; i < ; i++) {
tester.read();
}
}
} class writer implements Runnable {
private ReadWriteLockDemo tester = null; public writer(ReadWriteLockDemo tester) {
this.tester = tester;
} @Override
public void run() {
Calendar now = Calendar.getInstance();
System.out.println(now.getTime().getTime() + " W "
+ Thread.currentThread().getName() + " started");
for (int i = ; i < ; i++) {
tester.write();
}
}
}
上一篇:java数据库编程之DAO模式


下一篇:Source Insight、Xshell(putty)、Xftp