經(jīng)典線程同步問題

生產(chǎn)者消費者問題

http://en.wikipedia.org/wiki/Producer%E2%80%93consumer_problem

分別用鎖、信號量、同步監(jiān)視器模擬的例子。

package thread;

import java.util.Random;
import java.util.concurrent.Semaphore;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

class Item {
    int id;

    public Item(int id) {
        this.id = id;
    }

    public String toString() {
        return "[item " + id + "]";
    }
}

abstract class Buffer {
    int capacity;
    Item[] items;
    int count;
    int in, out;

    public Buffer(int capacity) {
        this.capacity = capacity;
        items = new Item[capacity];
        count = 0;
        in = out = 0;
    }

    abstract void put(Item item);

    abstract Item get();

    public void printBuf() {
        System.out.print("current buf status: [ ");
        for (int i = 0; i < capacity; i++) {
            System.out.print(items[i] + " ");
        }
        System.out.print("] ");
        System.out.print("count:" + count + " ");
        System.out.print("in:" + in + " out:" + out + " ");
        System.out.println();
    }
}

/**
 * 利用鎖實現(xiàn)線程同步的buffer
 * 
 * @author jd
 * 
 */
class LockBuffer extends Buffer {

    Lock lock = new ReentrantLock();
    Condition empty = lock.newCondition();
    Condition full = lock.newCondition();

    public LockBuffer(int capacity) {
        super(capacity);
    }

    public void put(Item item) {
        lock.lock();

        try {
            while (count == capacity)
                full.await();
            items[in] = item;
            in = (in + 1) % capacity;
            count++;
            empty.signal();

            System.out.println(Thread.currentThread().getName() + " put item " + item.id);
            printBuf();

        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        finally {
            lock.unlock();
        }
    }

    public Item get() {
        lock.lock();
        Item res = null;
        try {
            while (count == 0)
                empty.await();
            res = items[out];
            items[out] = null;
            out = (out + 1) % capacity;
            count--;
            full.signal();

            System.out.println(Thread.currentThread().getName() + " get item " + res.id);
            printBuf();

        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
        return res;
    }

}

/**
 * 利用信號量實現(xiàn)的線程同步的buffer
 * 
 * @author jd
 * 
 */
class SemaphoreBuffer extends Buffer {
    Semaphore mutex;
    Semaphore full;
    Semaphore empty;

    public SemaphoreBuffer(int capacity) {
        super(capacity);
        mutex = new Semaphore(1);
        full = new Semaphore(0);
        empty = new Semaphore(capacity);
    }

    public void put(Item item) {

        try {
            empty.acquire();
            mutex.acquire();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        items[in] = item;
        in = (in + 1) % capacity;
        count++;

        System.out.println(Thread.currentThread().getName() + " put item " + item.id);
        printBuf();

        mutex.release();
        full.release();

    }

    public Item get() {

        try {
            full.acquire();
            mutex.acquire();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        Item res = items[out];
        items[out] = null;
        out = (out + 1) % capacity;
        count--;

        System.out.println(Thread.currentThread().getName() + " get item " + res.id);
        printBuf();

        mutex.release();
        empty.release();
        return res;
    }

}

/**
 * 利用同步監(jiān)視器實現(xiàn)的線程同步的buffer
 * 
 * @author jd
 * 
 */
class MonitorBuffer extends Buffer {

    public MonitorBuffer(int capacity) {
        super(capacity);
    }

    public void put(Item item) {
        synchronized (this) {

            try {
                while (count == capacity)
                    wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            items[in] = item;
            in = (in + 1) % capacity;
            count++;
            notifyAll();

            System.out.println(Thread.currentThread().getName() + " put item " + item.id);
            printBuf();

        }
    }

    public Item get() {
        synchronized (this) {
            try {
                while (count == 0)
                    wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            Item res = items[out];
            items[out] = null;
            out = (out + 1) % capacity;
            count--;
            notifyAll();

            System.out.println(Thread.currentThread().getName() + " get item " + res.id);
            printBuf();
            return res;
        }
    }
}

class Producer implements Runnable {
    Buffer buf;
    Random rand = new Random();

    public Producer(Buffer buf) {
        this.buf = buf;
    }

    public void run() {
        for (int i = 0; i < 10; i++) {
            Item item = new Item(rand.nextInt(100));
            buf.put(item);

            try {
                Thread.sleep(500);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

    }
}

class Consumer implements Runnable {
    Buffer buf;

    public Consumer(Buffer buf) {
        this.buf = buf;
    }

    public void run() {
        for (int i = 0; i < 10; i++) {
            Item item = buf.get();

            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

public class BoundedBufferTest {

    public static void main(String[] args) {
        // Buffer buf = new LockBuffer(5);
        // Buffer buf = new SemaphoreBuffer(5);
        Buffer buf = new MonitorBuffer(5);

        // 3個生產(chǎn)者，3個消費者，每個生產(chǎn)或者消費10次。
        for (int i = 0; i < 3; i++) {
            new Thread(new Producer(buf), "p" + i).start();
            new Thread(new Consumer(buf), "c" + i).start();
        }

    }
}

View Code

讀者作家問題

http://en.wikipedia.org/wiki/Readers%E2%80%93writers_problem

模擬第一讀者問題。

package thread;

import java.util.concurrent.Semaphore;

/**
 * first reader writer problem, writer may starve;
 * 
 * @author jd
 * 
 */
class Article {
    String content = "The original content.";
    Semaphore mutex, wrt;
    int readCount;

    public Article() {
        mutex = new Semaphore(1);
        wrt = new Semaphore(1);
        readCount = 0;
    }

    public String read() throws InterruptedException {
        mutex.acquire();
        readCount++;
        if (readCount == 1)
            wrt.acquire();
        mutex.release();

        // reading is performed
        String res = content;
        System.out.println(Thread.currentThread().getName() + " is reading, the article is [" + res + "]");
        Thread.sleep(1000);

        mutex.acquire();
        readCount--;
        if (readCount == 0)
            wrt.release();
        mutex.release();

        return res;

    }

    public void Write(String str) throws InterruptedException {
        wrt.acquire();

        // writing is performed
        content = "new content(" + str + ")";
        System.out.println("content is changed to [" + content + "]");
        Thread.sleep(2000);

        wrt.release();
    }

}

class Reader implements Runnable {
    Article art;

    public Reader(Article a) {
        art = a;
    }

    public void run() {
        try {
            art.read();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

class Writer implements Runnable {
    Article art;

    public Writer(Article a) {
        art = a;
    }

    public void run() {
        try {
            art.Write(Thread.currentThread().getName());
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

public class ReaderWriterTest {

    public static void main(String[] args) {
        Article art = new Article();

        for (int i = 0; i < 3; i++) {
            new Thread(new Reader(art), "r" + i).start();
        }
        // writer starves;
        new Thread(new Writer(art), "w").start();

        for (int i = 0; i < 3; i++) {
            new Thread(new Reader(art), "rr" + i).start();
        }

    }
}

View Code

哲學(xué)家吃飯問題

http://en.wikipedia.org/wiki/Dining_philosophers_problem

模擬哲學(xué)家吃飯問題，該實現(xiàn)可能會產(chǎn)生死鎖。

package thread;

import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * This version may lead to deadlock.
 * 
 * @author jd
 * 
 */
class Chopstick {
    private Lock lock = new ReentrantLock();

    public void pickUp() {
        lock.lock();
    }

    public void putDown() {
        lock.unlock();
    }

}

class Philosopher extends Thread {
    private Chopstick left;
    private Chopstick right;

    public Philosopher(Chopstick left, Chopstick right) {
        this.left = left;
        this.right = right;
    }

    public void eat() {
        pickUp();
        // eating
        System.out.println(Thread.currentThread().getName() + " is eating");
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        putDown();
        System.out.println(Thread.currentThread().getName() + " done eating");

    }

    public void pickUp() {
        System.out.println(Thread.currentThread().getName() + " is trying to pick up left chopstick..");
        left.pickUp();

        // this sleep makes all philosopers pick up left chopstick and try to
        // pick up the right, which leads to a deadlock.
        // this rarely happens, but it can do.
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        System.out.println(Thread.currentThread().getName() + " is trying to pick up right chopstick..");
        right.pickUp();
    }

    public void putDown() {
        left.putDown();
        right.putDown();
    }

    public void run() {
        eat();
    }
}

public class DiningPhilosopherTest {

    public static void main(String[] args) {
        Chopstick[] chopsticks = new Chopstick[5];
        for (int i = 0; i < 5; i++)
            chopsticks[i] = new Chopstick();

        for (int i = 0; i < 5; i++) {
            new Philosopher(chopsticks[i], chopsticks[(i + 1) % 5]).start();
        }

    }
}

View Code

不死鎖版本，當(dāng)某個哲學(xué)家無法拿到兩只筷子時，會主動放棄已有資源（筷子）而放棄本次吃飯。

package thread;

import java.util.Random;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 
 * @author jd
 * 
 */
class ChopstickNoDeadLock {
    private Lock lock = new ReentrantLock();

    public boolean pickUp() {
        return lock.tryLock();
    }

    public void putDown() {
        lock.unlock();
    }

}

class PhilosopherNoDeadLock extends Thread {
    Random rand = new Random();
    private ChopstickNoDeadLock left;
    private ChopstickNoDeadLock right;

    public PhilosopherNoDeadLock(ChopstickNoDeadLock left, ChopstickNoDeadLock right) {
        this.left = left;
        this.right = right;
    }

    public void eat() {
        if (pickUp()) {
            pickUp();
            // eating
            System.out.println(Thread.currentThread().getName() + " is eating");
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            putDown();
            System.out.println(Thread.currentThread().getName() + " done eating");
        } else {
            System.out.println(Thread.currentThread().getName() + " gives up eating");

        }
    }

    public boolean pickUp() {
        System.out.println(Thread.currentThread().getName() + " is trying to pick up left chopstick..");
        if (!left.pickUp()) {
            return false;
        }

        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        System.out.println(Thread.currentThread().getName() + " is trying to pick up right chopstick..");
        if (!right.pickUp()) {
            left.putDown();
            return false;
        }

        return true;
    }

    public void putDown() {
        left.putDown();
        right.putDown();
    }

    public void run() {
        // for (int i = 0; i < 10; i++) {
        eat();
        // try {
        // Thread.sleep(rand.nextInt(10) * 100);
        // } catch (InterruptedException e) {
        // e.printStackTrace();
        // }
        // }
    }
}

public class DiningPhilosopherTestNoDeadLock {

    public static void main(String[] args) {
        ChopstickNoDeadLock[] chopsticks = new ChopstickNoDeadLock[5];
        for (int i = 0; i < 5; i++)
            chopsticks[i] = new ChopstickNoDeadLock();

        for (int i = 0; i < 5; i++) {
            new PhilosopherNoDeadLock(chopsticks[i], chopsticks[(i + 1) % 5]).start();
        }

    }
}

View Code

posted @ 2014-09-15 22:12 jdflyfly 閱讀(320) 評論(1) 收藏舉報

刷新頁面返回頂部

jdflyfly

經(jīng)典線程同步問題

公告