Java多线程模式（二）

Guarded Suspension Pattern

     该模式描述的是当一个线程在执行某个操作时，但由于其他资源还没有准备好，需要等待，那么就等待资源准备好才开始自己的操作。我们直接看代码例子：

public class Request {

    private Stringname;

    public Request(String name) {

        this.name = name;

    }

    public String getName() {

        returnname;

    }

    @Override

    public String toString() {

        return"[ Request " +name +" ]";

    }

}

public class RequestQueue {

    final private LinkedList<Request>queue = new LinkedList<Request>();

    public synchronizedvoid putRequest(Request request) {

        this.queue.addLast(request);

        notifyAll();

    }

    publicsynchronized Request getRequest() {

       // 多线程版本的if

        while (this.queue.size() <= 0) {

           try {

                wait();

            }catch (InterruptedException e) {

            }

        }

        return queue.removeFirst();

    }

}

import java.util.Random;

public class ClientThreadextends Thread {

    private Random      random;

    private RequestQueuerequestQueue;

    public ClientThread(RequestQueue requestQueue, String name,long seed) {

        super(name);

        this.requestQueue = requestQueue;

        this.random =new Random(seed);

    }

    @Override

    public void run() {

        for (int i = 0; i < 10000; i++) {

            Request request = new Request("No." + i);

            System.out.println(Thread.currentThread().getName() +" requests " + request);

            this.requestQueue.putRequest(request);

            try {

                Thread.sleep(this.random.nextInt(1000));

            } catch (InterruptedException e) {

                e.printStackTrace();

            }

        }

    }

}

import java.util.Random;

public class ServerThreadextends Thread {

    private Random      random;

    private RequestQueuequeue;

    public ServerThread(RequestQueue queue, String name,long seed) {

        super(name);

        this.queue = queue;

        random =new Random(seed);

    }

    @Override

    public void run() {

        for (int i = 0; i < 10000; i++) {

            Request request = queue.getRequest();

            System.out.println(Thread.currentThread().getName() +" handles " + request);

            try {

                Thread.sleep(random.nextInt(1000));

            } catch (InterruptedException e) {

            }

        }

    }

}

publicclass Main {

    public static void main(String[] args) {

        RequestQueue queue = new RequestQueue();

        ServerThread serverThread = new ServerThread(queue,"ServerThread", 3141592l);

        ClientThread clientThread = new ClientThread(queue,"ClientThread", 6535897l);

        serverThread.start();

        clientThread.start();

    }

}

这段代码的关键在ReqeustQueue类的getReqeust()方法，在该方法中，判断队列是否小于或等于0，如果是，那么就等待队列有数据之后在进行获取Request对象的操作，注意这里使用的是while，而非if。Single Threaded Execution Pattern  只有一个线程可以进入临界区，其他线程不能进入，进行等待；而Guarded Suspension Pattern中，线程要不要等待，由警戒条件决定。只有RequestQueue类使用到了wait/notifyAll，Guarded Suspension Pattern的实现是封闭在RequestQueue类里的。

Balking Pattern

该模式的重点是，如果一个请求的资源状态还没有准备好，那么就不进行处理，直接返回，它与Guarded Suspension Pattern的区别在于Guarded Suspension Pattern在警戒条件不成立时，线程等待，而Balking Pattern线程直接返回。我们来看代码实现：

import java.io.File;

import java.io.FileWriter;

import java.io.IOException;

publicclass Data {

    private final Stringfilename;

    private String      content;

    privateboolean     changed;

    public Data(String filename, String content) {

        this.filename = filename;

        this.content = content;

        this.changed =true;

    }

    public synchronizedvoid change(String content) {

        this.content = content;

        this.changed =true;

    }

    publicsynchronizedvoid save() {

        while (!this.changed) {

            return;

        }

        doSave();

        this.changed =false;

    }

    private void doSave() {

        System.out.println(Thread.currentThread().getName() +"calls doSave, content = "

                + this.content);

        File file = new File(filename);

        FileWriter writer = null;

        try {

            writer = new FileWriter(file, true);

            writer.write(this.content);

        } catch (IOException e) {

        } finally {

            if (writer !=null) {

                try {

                    writer.close();

                } catch (IOException e) {

                    e.printStackTrace();

                }

            }

        }

    }

}

 

import java.util.Random;

public class ChangerThreadextends Thread {

    private Data  data;

    private Randomrandom =new Random();

    public ChangerThread(String name, Data data) {

        super(name);

        this.data = data;

    }

    @Override

    public void run() {

        int i = 0;

        while (true) {

            i++;

            String content = "No." + i;

            this.data.change(content);

            try {

                Thread.sleep(random.nextInt(1000));

            } catch (InterruptedException e) {

            }

            this.data.save();

        }

    }

}

 

import java.util.Random;

public class SaverThreadextends Thread {

    private Data  data;

    private Randomrandom =new Random();

    public SaverThread(String name, Data data) {

        super(name);

        this.data = data;

    }

    @Override

    public void run() {

        while (true) {

            this.data.save();

            try {

                Thread.sleep(this.random.nextInt(1000));

            } catch (InterruptedException e) {

            }

        }

    }

    public static void main(String[] args) {

        Data data = new Data("data.txt","(empty)");

        new SaverThread("SaverThread", data).start();

        new ChangerThread("ChangerThread", data).start();

    }

}

Producer-Consumer Pattern

该模式即经典的生产-消费模式。该模式在生产者和消费者之间加入一个“桥梁参与者”，以这个参与者来缓冲线程的处理速度之差。一般会有多个生产者和多个消费者。

import java.io.Serializable;

public class Data implements Serializable {

    /**

     * 

     */

    private static final long serialVersionUID = 7212370995222659529L;

    private String            name;

    public Data(String name) {

        this.name = name;

    }

    @Override

    public String toString() {

        return"[ Data name = " +this.name +" ]";

    }

}

 

import java.util.LinkedList;

/**

 * 数据传输channel，默认大小100，可以通过构造函数定制channel的大小。channel为FIFO模型

 */

public class Channel {

    private final LinkedList<Data>buffer     =new LinkedList<Data>();

    private int                   bufferSize = 100;

    public Channel() {

        super();

    }

    public Channel(int channelSize) {

        this.bufferSize = channelSize;

    }

    /**

     * put数据到channel中，当channel的buffer大小大于或等于指定大小时，方法将进行等待

     * 

     * @param data

     */

    public synchronizedvoid put(Data data) {

        while (buffer.size() >=this.bufferSize) {

            try {

                wait();

            } catch (InterruptedException e) {

            }

        }

        this.buffer.addLast(data);

        System.out.println(Thread.currentThread().getName() +" put data " + data);

        notifyAll();

    }

    /**

     * 从channel中获取数据，当channel中没有数据时，进行等待

     * 

     * @return

     */

    public synchronized Data take() {

        while (this.buffer.size() == 0) {

            try {

                wait();

            } catch (InterruptedException e) {

            }

        }

        Data data = this.buffer.removeFirst();

        System.out.println(Thread.currentThread().getName() +" take date " + data);

        notifyAll();

        return data;

    }

}

 

import java.util.Random;

public class ComsumerThreadextends Thread {

    private Channel channel;

    private Random  random =new Random();

    public ComsumerThread(String name, Channel channel) {

        super(name);

        this.channel = channel;

    }

    @Override

    public void run() {

        while (true) {

            this.channel.take();

            try {

                Thread.sleep(random.nextInt(1000));

            } catch (InterruptedException e) {

            }

        }

    }

}

 

import java.util.Random;

public class ProducerThreadextends Thread {

    private Channel    channel;

    private Random     random =new Random();

    privatestaticintdataNo = 0;

    public ProducerThread(String name, Channel channel) {

        super(name);

        this.channel = channel;

    }

    @Override

    public void run() {

        while (true) {

            Data data = new Data("No." + nextDataNo());

            this.channel.put(data);

            try {

                Thread.sleep(random.nextInt(1000));

            } catch (InterruptedException e) {

            }

        }

    }

    publicstaticsynchronizedint nextDataNo() {

        return ++dataNo;

    }

}

 

public class MainThread {

    public static void main(String[] args) {

        int channelSize = 1000;

        Channel channel = new Channel(channelSize);

        ProducerThread producer1 = new ProducerThread("Producer1", channel);

        ProducerThread producer2 = new ProducerThread("Producer2", channel);

        ComsumerThread comsumer1 = new ComsumerThread("Comsumer1", channel);

        ComsumerThread comsumer2 = new ComsumerThread("Comsumer2", channel);

        ComsumerThread comsumer3 = new ComsumerThread("Comsumer3", channel);

        producer1.start();

        producer2.start();

        comsumer1.start();

        comsumer2.start();

        comsumer3.start();

    }

}