03.Java多线程并发库API使用2

时间:2023-03-08 15:34:19

1.多个线程之间共享数据的方式探讨

1、如果每个线程执行的代码相同,可以使用同一个Runnable对象,这个Runnable对象中有那个共享数据,例如,买票系统就可以这么做。

2、如果每个线程执行的代码不同,这时候需要用不同的Runnable对象,有如下两种方式来实现这些Runnable对象之间的数据共享:

  • 将共享数据封装在另外一个对象中,然后将这个对象逐一传递给各个Runnable对象。每个线程对共享数据的操作方法也分配到那个对象身上去完成,这样容易实现针对该数据进行的各个操作的互斥和通信。
  • 将这些Runnable对象作为某一个类中的内部类,共享数据作为这个外部类中的成员变量,每个线程对共享数据的操作方法也分配给外部类,以便实现对共享数据进行的各个操作的互斥和通信,作为内部类的各个Runnable对象调用外部类的这些方法。

上面两种方式的组合:将共享数据封装在另外一个对象中,每个线程对共享数据的操作方法也分配到那个对象身上去完成,对象作为这个外部类中的成员变量或方法中的局部变量,每个线程的Runnable对象作为外部类中的成员内部类或局部内部类。

总之,要同步互斥的几段代码最好是分别放在几个独立的方法中,这些方法再放在同一个类中,这样比较容易实现它们之间的同步互斥和通信。

极端且简单的方式,即在任意一个类中定义一个static的变量,这将被所有线程共享。

示例代码

 package com.chunjiangchao.thread;
/**
* 多线程之间数据共享
* @author chunjiangchao
*
*/
public class MultiThreadShareDataDemo { public static void main(String[] args) {
Data data = new Data();
new Thread(new IncrementRunnable(data)).start();
new Thread(new DecrementtRunnable(data)).start(); final Data data2 = new Data();
new Thread(new Runnable() { @Override
public void run() {
data2.increment();
}
}).start();
new Thread(new Runnable() { @Override
public void run() {
data2.decrement();
}
}).start();
}
//对共享数据进行增加
private static class IncrementRunnable implements Runnable{
private Data data ;
public IncrementRunnable(Data data){
this.data = data;
}
public void run() {
data.increment();
}
}
//对共享数据进行减少
private static class DecrementtRunnable implements Runnable{
private Data data ;
public DecrementtRunnable(Data data){
this.data = data;
}
public void run() {
data.decrement();
}
} //共享数据
private static class Data{
private int temp=0;
public synchronized void increment(){
temp++;
System.out.println(Thread.currentThread()+"中temp的值为:"+temp);
}
public synchronized void decrement(){
temp--;
System.out.println(Thread.currentThread()+"中temp的值为:"+temp);
}
} }

2.java5线程并发库的应用(Executors)

static ExecutorService newFixedThreadPool(int nThreads) 创建一个可重用固定线程数的线程池,以共享的*队列方式来运行这些线程。(创建固定线程池)

如果在这个线程池里面,创建的线程为3个线程,但是交给的任务时10个任务的话,那么,线程池里面的线程就会运行完3个线程后,接着运行3个线程,直到所有的线程运行完毕。

List<Runnable> shutdownNow()试图停止所有正在执行的活动任务,暂停处理正在等待的任务,并返回等待执行的任务列表。

shutdown()启动一次顺序关闭,执行以前提交的任务,但不接受新任务。

static ExecutorService newCachedThreadPool()创建一个可根据需要创建新线程的线程池,但是在以前构造的线程可用时将重用它们。(动态创建线程池,有多少任务,自动创建多少线程)

static ExecutorService  newSingleThreadExecutor():创建单个线程,如果线程死掉了,它会自动找个替补线程补上去。(如何实现线程死掉之后重新启动)?

static ScheduledExecutorService newScheduledThreadPool(int corePoolSize):创建一个定时线程池

实例代码:

 package com.chunjiangchao.thread;

 import java.util.Date;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit; /**
* 线程并发库,线程池的使用
* @author chunjiangchao
*
*/
public class ExecuterDemo { public static void main(String[] args) {
// ExecutorService threadPool = Executors.newFixedThreadPool(3);//开了固定的三个线程
// ExecutorService threadPool = Executors.newCachedThreadPool();//开了10个线程
ExecutorService threadPool = Executors.newSingleThreadExecutor();//开了一个固定的线程
for(int i=0;i<10;i++){
final int loop = i;
threadPool.execute(new Runnable(){
public void run() {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
// e.printStackTrace();
}
System.out.println(Thread.currentThread().getName()+" outer "+loop);
} });
}
/*
shutdownNow执行的结果为:
pool-1-thread-3 outer
pool-1-thread-1 outer
pool-1-thread-2 outer * */
// threadPool.shutdownNow();
/*shutdown会执行完所有已经提交的任务,不会处理shutdown后提交的任务,而且在后面提交Runnable的时候,
* 会抛出异常java.util.concurrent.RejectedExecutionException*/
threadPool.shutdown();
// threadPool.execute(new Runnable(){
//
// @Override
// public void run() {
// System.out.println("不会进行处理");
// }
//
// });
//实现定时器效果
Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(new Runnable(){ @Override
public void run() {
System.out.println("执行定时器结果"+new Date().toLocaleString());
} }, 2, 4, TimeUnit.SECONDS);//每隔4s玩一次
} }

3.Callable&Future

Future取得的结果类型和Callable返回的结果类型必须一致,这是通过泛型来实现的。

Callable要采用ExecutorSevice的submit方法提交,返回的future对象可以取消任务。

CompletionService用于提交一组Callable任务,其take方法返回已完成的一个Callable任务对应的Future对象。

take()   获取并移除表示下一个已完成任务的 Future,如果目前不存在这样的任务,则等待

示例代码

 package com.chunjiangchao.thread;

 import java.util.Date;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future; /**
* Callable&Future的使用
* @author chunjiangchao
*
*/
public class CallableAndFutureDemo { public static void main(String[] args) {
ExecutorService newSingleThreadExecutor = Executors.newSingleThreadExecutor();
//提交单一任务
Future<String> submit = newSingleThreadExecutor.submit(new Callable<String>(){ @Override
public String call() throws Exception {
printTime();
mSleep(3000);
printTime();
return "我这有返回值,你看看是不是";
} });
mSleep(500);
try {
String string = submit.get();
System.out.println(string);
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}
// submit.cancel(true);//可以对任务进行取消
//提交多个任务
Executor executor = Executors.newCachedThreadPool();
ExecutorCompletionService<String> completionService = new ExecutorCompletionService<>(executor);
for(int i=0;i<10;i++){
final int loop = i;
completionService.submit(new Callable<String>(){ @Override
public String call() throws Exception {
mSleep(1000*loop);
return "提交多任务有返回结果"+loop;
} });
}
for(int i=0;i<10;i++){
try {
Future<String> result = completionService.take();
printTime();
System.out.println(result.get());
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
}
/*
* 打印 结果如下
2016-4-18 11:57:46
2016-4-18 11:57:49
我这有返回值,你看看是不是
2016-4-18 11:57:49
提交多任务有返回结果0
2016-4-18 11:57:50
提交多任务有返回结果1
2016-4-18 11:57:51
提交多任务有返回结果2
2016-4-18 11:57:52
提交多任务有返回结果3
2016-4-18 11:57:53
提交多任务有返回结果4
2016-4-18 11:57:54
提交多任务有返回结果5
2016-4-18 11:57:55
提交多任务有返回结果6
2016-4-18 11:57:56
提交多任务有返回结果7
2016-4-18 11:57:57
提交多任务有返回结果8
2016-4-18 11:57:58
提交多任务有返回结果9
*/ }
private static void mSleep(long time){
try {
Thread.sleep(time);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
private static void printTime(){
System.out.println(new Date().toLocaleString());
} }

4.java5的线程锁技术

Lock的使用

 package com.chunjiangchao.thread;

 import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock; /**
* lock的使用
*/
public class LockDemo { public static void main(String[] args) {
final Outputer outputer = new Outputer();
for(int index=0;index<10;index++){
final int loop = index;
new Thread(new Runnable() {
public void run() {
// outputer.print("chunjiangchao"+loop);
outputer.synPrint("chunjiangchao"+loop);
}
}).start(); }
}
private static class Outputer{
private Lock lock = new ReentrantLock();
public void print(String name){
int length = name.length();
lock.lock();
try {
for(int i=0;i<length;i++){
Thread.sleep(100);
System.out.print(name.charAt(i)+" ");
}
System.out.println();
} catch (Exception e) {
e.printStackTrace();
}finally{
lock.unlock();
} }
/**
* 同步代码块的作用,和上面添加Lock锁的作用相同,只不过锁的对象不一样而已
* @param name
*/
public synchronized void synPrint(String name){
int length = name.length();
try {
for(int i=0;i<length;i++){
Thread.sleep(100);
System.out.print(name.charAt(i)+" ");
}
System.out.println();
} catch (Exception e) {
e.printStackTrace();
}
}
} }

5.java5读写锁技术的妙用(ReadWriteLock)

Lock比传统线程模型中的synchronized方式更加面向对象,与生活中的锁类似,锁本身也应该是一个对象。两个线程执行的代码片段要实现同步互斥的效果,它们必须用同一个Lock对象。Lock lock= new ReentrantLock( )

ReadWriteLock  rwl = new ReentrantReadWriteLock( )

读写锁:分为读锁和写锁,多个读锁不互斥,读锁与写锁互斥,这是由jvm自己控制的,你只要上好相应的锁即可。如果你的代码只读数据,可以很多人同时读,但不能同时写,那就上读锁;如果你的代码修改数据,只能有一个人在写,且不能同时读取,那就上写锁。总之,读的时候上读锁,写的时候上写锁!

示例代码(读锁与读锁并发,写锁与写锁并发,读锁与写锁互斥)

 package com.chunjiangchao.thread;

 import java.util.Random;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
/**
* 使用读写锁 查看打印结果发现读锁与读锁之间并发,写锁与写锁间并发,读与写之间是互斥的
* @author chunjaingchao
*
*/
public class ReadWriteLockDemo {
public static void main(String[] args) {
final Queue q3 = new Queue();
for(int i=0;i<3;i++)
{
new Thread(){
public void run(){
while(true){
q3.get();
}
}
}.start();
new Thread(){
public void run(){
while(true){
q3.put(new Random().nextInt(10000));
}
}
}.start();
}
}
static class Queue{
private Integer integer = null;//共享数据,只能有一个线程能写该数据,但可以有多个线程同时读该数据。
ReadWriteLock rwl = new ReentrantReadWriteLock();
public void get(){
rwl.readLock().lock();
try {
System.out.println(Thread.currentThread().getName() + "*****读取******");
Thread.sleep(200);
System.out.println(Thread.currentThread().getName() + "******读取*****" + integer);
} catch (InterruptedException e) {
e.printStackTrace();
}finally{
rwl.readLock().unlock();
}
}
public void put(Integer data){
rwl.writeLock().lock();
try {
System.out.println(Thread.currentThread().getName() + "######写数据#######");
Thread.sleep(200);
this.integer = data;
System.out.println(Thread.currentThread().getName() + "#######写数据#######" + data);
} catch (InterruptedException e) {
e.printStackTrace();
}finally{
rwl.writeLock().unlock();
}
}
}
}

在线程操作某个方法,执行这个方法的时候。

自己挂写锁,然后自己挂读锁也是可以的(因为这是在当前线程同一个方法中的)。自己挂写锁,是为了防止其他人进入程序进行写的操作。但是,不应该进制自己进入。(在Hibernate中,锁分为读锁、写锁、更新锁)

在JDKAPI中有相关的实例代码如下

 class CachedData {
Object data;
volatile boolean cacheValid;
ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();
void processCachedData() {
rwl.readLock().lock();//添加读锁
if (!cacheValid) {
// Must release read lock before acquiring write lock
rwl.readLock().unlock();//如果没有数据,将读锁释放
rwl.writeLock().lock();//添加写锁
// Recheck state because another thread might have acquired
// write lock and changed state before we did.
if (!cacheValid) {
data = ...
cacheValid = true;
}
// Downgrade by acquiring read lock before releasing write lock
rwl.readLock().lock();//添加读锁
rwl.writeLock().unlock(); // Unlock write, still hold read//释放写锁
}
use(data);
rwl.readLock().unlock();//释放读锁
}
}

问题:设计缓存系统

缓存系统的概念:你要找数据不要直接去找数据库,可以直接找我。 我如果没有,查找数据库给你。与你直接查找是一样的。好处就是下一次你再来的时候,我就不用操作数据库了。我直接给你。

6.java5条件阻塞Condition的应用

在等待 Condition 时,允许发生“虚假唤醒”,这通常作为对基础平台语义的让步。对于大多数应用程序,这带来的实际影响很小,因为 Condition 应该总是在一个循环中被等待,并测试正被等待的状态声明。某个实现可以随意移除可能的虚假唤醒,但建议应用程序程序员总是假定这些虚假唤醒可能发生,因此总是在一个循环中等待。(记住:每次在等待的时候,都要将判断放在while循环中,防止伪唤醒出现

一个锁内部可以有多个Condition,即有多路等待和通知,可以参看jdk1.5提供的Lock与Condition实现的可阻塞队列的应用案例,从中除了要体味算法,还要体味面向对象的封装。在传统的线程机制中一个监视器对象上只能有一路等待和通知,要想实现多路等待和通知,必须嵌套使用多个同步监视器对象。(如果只用一个Condition,两个放的都在等,一旦一个放的进去了,那么它通知可能会导致另一个放接着往下走。)

问题:此处为什么要创建两个Condition对象?只创建一个不就行了?

答:如果本道程序只有两个线程的话,只创建一个Condition对象就行了。如果是超过4个线程。例如两个存放线程、两个读取线程。如果你只创建一个Condition对象,在signal的时候,会唤醒所有都处在等待状态的线程。而不是针对某一种类型的线程。(没有针对性)

在API文档中有如下实例代码

 class BoundedBuffer {
final Lock lock = new ReentrantLock();
final Condition notFull = lock.newCondition();
final Condition notEmpty = lock.newCondition();
final Object[] items = new Object[100];
int putptr, takeptr, count;
public void put(Object x) throws InterruptedException {
lock.lock();
try {
while (count == items.length)
notFull.await();
items[putptr] = x;
if (++putptr == items.length) putptr = 0;
++count;
notEmpty.signal();
} finally {
lock.unlock();
}
}
public Object take() throws InterruptedException {
lock.lock();
try {
while (count == 0)
notEmpty.await();
Object x = items[takeptr];
if (++takeptr == items.length) takeptr = 0;
--count;
notFull.signal();
return x;
} finally {
lock.unlock();
}
}
}

利用BoundedBuffer写一个简单是生产者消费者模式

 public class BoundedBufferDemo {

     public static void main(String[] args) {
final BoundedBuffer boundedBuffer = new BoundedBuffer();
new Thread(new Runnable(){ @Override
public void run() {
while(true){
try {
Thread.sleep(1000);
int nextInt = new Random().nextInt();
System.out.println(new Date().toLocaleString()+"存放数据"+nextInt);
boundedBuffer.put(nextInt);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
} }).start();
new Thread(new Runnable(){ @Override
public void run() {
while(true){
try {
Thread.sleep(new Random().nextInt(1000));
System.out.println(new Date().toLocaleString()+"获取数据"+boundedBuffer.take());
} catch (InterruptedException e) {
e.printStackTrace();
}
}
} }).start();
}
}

问题:子线程循环10次,接着主线程循环100,接着又回到子线程循环10次,接着再回到主线程又循环100,如此循环50次,请写出程序。

 package com.chunjiangchao.thread;

 import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock; /**
* 类似于生产者消费者
* 子线程循环10次,接着主线程循环100,接着又回到子线程循环10次,接着再回到主线程又循环100,如此循环50次,请写出程序。
* @author chunjiangchao
*/
public class ConditionDemo { public static void main(String[] args) {
final Business business = new Business();
new Thread(new Runnable() {
public void run() {
for(int i = 0;i<10;i++){
business.sub(i);
}
}
}).start();
new Thread(new Runnable() {
public void run() {
for(int i = 0;i<10;i++){
business.main(i);
}
}
}).start();
}
private static class Business{
private Lock lock = new ReentrantLock();
private Condition condition = lock.newCondition();
private boolean bShouldSub = true;
public void main(int loop){
lock.lock();
while(bShouldSub){
try {
condition.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
for(int i=0;i<100;i++){
try {
Thread.sleep(10);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(loop+"……main……"+i);
}
bShouldSub = true;
condition.signal();
lock.unlock();//应该写在finally代码块里面
}
public void sub(int loop){
lock.lock();
while(!bShouldSub){
try {
condition.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
for(int i=0;i<10;i++){
try {
Thread.sleep(10);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(loop+"……sub……"+i);
}
bShouldSub = false;
condition.signal();
lock.unlock();//应该写在finally代码块里
} }
}

问题:怎样实现3个线程的交互通信?

 package com.chunjiangchao.thread;

 import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock; /**
* 多线程之间的通信
* 三个线程交互执行 A-B-C-A-B-C
* @author chunjiangchao
*
*/
public class ThreeConditionDemo { public static void main(String[] args) {
final Business business = new Business();
new Thread(new Runnable() {
public void run() {
for(int i = 0;i<10;i++){
business.one(i);
}
}
}).start();
new Thread(new Runnable() {
public void run() {
for(int i = 0;i<10;i++){
business.two(i);
}
}
}).start();
new Thread(new Runnable() {
public void run() {
for(int i = 0;i<10;i++){
business.three(i);
}
}
}).start();
}
private static class Business{
private Lock lock = new ReentrantLock();
private Condition condition1 = lock.newCondition();
private Condition condition2 = lock.newCondition();
private Condition condition3 = lock.newCondition();
private int whichOne = 1;
public void one(int loop){
try {
lock.lock();
while(whichOne!=1){
condition1.await();
}
for(int i=0;i<10;i++){
Thread.sleep(10);
System.out.println("one "+loop+" 当前执行 "+i);
}
whichOne = 2;
condition2.signal();
} catch (Exception e) {
e.printStackTrace();
}finally{
lock.unlock();
}
}
public void two(int loop){
try {
lock.lock();
while(whichOne!=2){
condition2.await();
}
for(int i=0;i<10;i++){
Thread.sleep(10);
System.out.println("two "+loop+" 当前执行 "+i);
}
whichOne = 3;
condition3.signal();
} catch (Exception e) {
e.printStackTrace();
}finally{
lock.unlock();
}
}
public void three(int loop){
try {
lock.lock();
while(whichOne!=3){
condition3.await();
}
for(int i=0;i<10;i++){
Thread.sleep(10);
System.out.println("three "+loop+" 当前执行 "+i);
}
whichOne = 1;
condition1.signal();
} catch (Exception e) {
e.printStackTrace();
}finally{
lock.unlock();
}
}
} }

未完待续……