浅谈Java多线程（二）

7-21 后端 magicwt 2,531 views

浅谈Java多线程（二）

7-21 2,531 views

线程的协调

在Java多线程中，对于线程之间共享的实例资源，可以通过synchronized修饰符修饰其方法，实现线程之间的同步。另外，在多线程设计中，还需考虑到线程之间的协调。关于协调的一个典型设计模式便是Producer-Consumer（生产者-消费者）模式。

Producer-Consumer（生产者-消费者）模式

在这一模式中，存在Producer和Consumer两类线程，Producer线程用于生成Data（共享数据资源），而Consumer线程用于消费Data，在Producer和Consumer之间，存在对于Data生成和消费的协调，即当不存在Data时，Consumer线程需要等待Producer线程生成新的Data，而当Data过多时，Producer线程需要等待Consumer线程消费过多的Data。在Producer-Consumer模式中，引入了Channel（管道）类，负责Data在各线程之间的协调。Producer-Consumer模式的UML类图如下所示。

在上图中，Producer线程类和Consumer线程类均包含对Channel类对象的引用，而Channel类对象封装了Data类，并分别实现了生产和消费Data的同步方法produce和consume。在produce和consume方法中，通过使用wait和notifyAll方法进一步实现Data的协调。

wait、notify、notifyAll

在Java中，wait、notify、notifyAll是Object类的方法，用于实现对调用对象方法的线程的暂停和唤醒。wait用于暂停线程，将其放入对象的wait set（线程等待集合），而notify、notifyAll方法用于唤醒wait set中的线程，使其继续执行。notify和notifyAll的不同是，当wait set中存在多个线程时，notify只会从中随机唤醒一个线程，而notifyAll会从中唤醒所有线程，由其进行竞争，获得同步锁并继续执行。
一个关于wait、notify、notifyAll使用的简单示例如图所示。

Producer-Consumer（生产者-消费者）模式的实现示例

Producer和Consumer线程类

package com.wt.pc;  
 
public class Producer extends Thread{  
 
    //对Channel对象的引用  
    Channel channel = null;  
      
    //Consumer类构造函数  
    public Producer(String producerName,Channel channel)  
    {  
        super(producerName);  
        this.channel = channel;  
    }  
      
    //Producer线程每隔500ms尝试生成新的data  
    public void run()  
    {  
        try{  
            while(true)  
            {  
                channel.produce(Integer.toString(Channel.dataId++));  
                Thread.sleep(500);  
            }  
        }catch(Exception e){}  
          
    }  
}

package com.wt.pc;

public class Producer extends Thread{

//对Channel对象的引用

Channel channel = null;

//Consumer类构造函数

public Producer(String producerName,Channel channel)

{

super(producerName);

this.channel = channel;

}

//Producer线程每隔500ms尝试生成新的data

public void run()

{

try{

while(true)

{

channel.produce(Integer.toString(Channel.dataId++));

Thread.sleep(500);

}

}catch(Exception e){}

}

package com.wt.pc;  
 
public class Consumer extends Thread{  
 
    //对Channel对象的引用  
    Channel channel = null;  
      
    //Consumer类构造函数  
    public Consumer(String consumerName,Channel channel)  
    {  
        super(consumerName);  
        this.channel = channel;  
    }  
      
    //Consumer线程每隔500ms尝试消费data  
    public void run()  
    {  
        try{  
            while(true)  
            {  
                channel.consume();  
                Thread.sleep(500);  
            }  
        }catch(Exception e){}  
          
    }  
 
}

package com.wt.pc;

public class Consumer extends Thread{

//对Channel对象的引用

Channel channel = null;

//Consumer类构造函数

public Consumer(String consumerName,Channel channel)

{

super(consumerName);

this.channel = channel;

}

//Consumer线程每隔500ms尝试消费data

public void run()

{

try{

while(true)

{

channel.consume();

Thread.sleep(500);

}

}catch(Exception e){}

}

Channel和Data类

package com.wt.pc;  
 
public class Channel {  
    //静态变量，用于生成data名称  
    static int dataId = 0;  
      
    //存储data的数组  
    Data dataList[];  
    //当前未消费data的头序号  
    int head;  
    //当前未消费data的尾序号的下一个  
    int tail;  
    //当前未消费data的数目  
    int count;  
      
    //Channel类构造函数  
    //数组容量为3，其他值初始化为0  
    public Channel()  
    {  
        dataList = new Data[3];  
        head = 0;  
        tail = 0;  
        count = 0;  
    }  
 
    //produce方法，用于生成data  
    public synchronized void produce(String dataName)throws Exception{  
        //当数组容量已满时，即不能再生成新data时，当前线程进入wait set  
        while (count>=3)  
        {  
            wait();  
        }  
        //生成新的data  
        System.out.println(Thread.currentThread().getName()+" is producing "+dataName);  
        dataList[tail] = new Data();  
        dataList[tail].setDataName(dataName);  
        tail = (tail+1)%3;  
        count++;  
        Thread.sleep(400);  
        //唤醒wait set中的线程  
        notifyAll();  
    }  
      
    //consume方法，用于消费data  
    public synchronized void consume()throws Exception{  
        //当数组容量为空时，即不能再消费data时，当前线程进入wait set  
        while (count<=0)  
        {  
            wait();  
        }  
        //消费data  
        System.out.println(Thread.currentThread().getName()+" is consuming "+dataList[head].getDataName());  
        head = (head+1)%3;  
        count--;  
        Thread.sleep(300);  
        //唤醒wait set中的线程  
        notifyAll();          
    }     
}

package com.wt.pc;

public class Channel {

//静态变量，用于生成data名称

static int dataId = 0;

//存储data的数组

Data dataList[];

//当前未消费data的头序号

int head;

//当前未消费data的尾序号的下一个

int tail;

//当前未消费data的数目

int count;

//Channel类构造函数

//数组容量为3，其他值初始化为0

public Channel()

{

dataList = new Data[3];

head = 0;

tail = 0;

count = 0;

}

//produce方法，用于生成data

public synchronized void produce(String dataName)throws Exception{

//当数组容量已满时，即不能再生成新data时，当前线程进入wait set

while (count>=3)

{

wait();

}

//生成新的data

System.out.println(Thread.currentThread().getName()+" is producing "+dataName);

dataList[tail] = new Data();

dataList[tail].setDataName(dataName);

tail = (tail+1)%3;

count++;

Thread.sleep(400);

//唤醒wait set中的线程

notifyAll();

}

//consume方法，用于消费data

public synchronized void consume()throws Exception{

//当数组容量为空时，即不能再消费data时，当前线程进入wait set

while (count<=0)

{

wait();

}

//消费data

System.out.println(Thread.currentThread().getName()+" is consuming "+dataList[head].getDataName());

head = (head+1)%3;

count--;

Thread.sleep(300);

//唤醒wait set中的线程

notifyAll();

}

package com.wt.pc;  
public class Data {  
    String dataName;  
 
    public String getDataName() {  
        return dataName;  
    }  
 
    public void setDataName(String dataName) {  
        this.dataName = dataName;  
    }  
}

package com.wt.pc;

public class Data {

String dataName;

public String getDataName() {

return dataName;

}

public void setDataName(String dataName) {

this.dataName = dataName;

}

主函数

public static void main(String args[])  
{  
    Channel channel = new Channel();  
    new Producer("p1",channel).start();  
    new Producer("p2",channel).start();  
    new Producer("p3",channel).start();  
    new Consumer("c1",channel).start();  
    new Consumer("c2",channel).start();  
    new Consumer("c3",channel).start();  
      
}

public static void main(String args[])

{

Channel channel = new Channel();

new Producer("p1",channel).start();

new Producer("p2",channel).start();

new Producer("p3",channel).start();

new Consumer("c1",channel).start();

new Consumer("c2",channel).start();

new Consumer("c3",channel).start();

}

执行结果

p2 is producing 0
c2 is consuming 0
p3 is producing 2
p1 is producing 1
c3 is consuming 2
c1 is consuming 1
p1 is producing 5
p3 is producing 4
c2 is consuming 5
p2 is producing 3
p3 is producing 7
c1 is consuming 4
c3 is consuming 3
p3 is producing 9
p1 is producing 6
c2 is consuming 7
p2 is producing 8
c3 is consuming 9
c1 is consuming 6
p2 is producing 12
p3 is producing 10
c2 is consuming 8
……

从执行结果中可以看出，生成Data线程的执行次数p和消费Data线程的执行次数c始终满足：

p>=c且p<=c+3

即保证Data数组在消费时存在Data但也不超过数组容量，从而有效实现Producer和Consumer线程类关于Data的协调。

总结

在Java多线程设计中，需要充分考虑线程之间的同步和协调。针对不同的应用场景，可以采用不同的设计模式，已有的设计模式有Single Threaded Execution、Immutable、Guarded Suspension、Balking、Producer-Consumer、Read-Write Lock、Thread-Per-Message、Worker Thread等，具体可进一步参考网上有关“Java多线程设计模式”的教程。

版权属于: 我爱我家

原文地址: http://magicwt.com/2012/07/21/%e6%b5%85%e8%b0%88java%e5%a4%9a%e7%ba%bf%e7%a8%8b%ef%bc%88%e4%ba%8c%ef%bc%89/

转载时必须以链接形式注明原始出处及本声明。