java 并发(4) 读者写者问题的几种实现方法及性能比较

读者-写者问题

1、读不阻塞其他读

2、读阻塞其他写

3、写阻塞其他读与其他写

 

问题分为三种：读者优先、写者优先、公平竞争

 

读者优先：如果当前运行为读线程，则后续的读线程可以不阻塞，直接读

                 如果当前运行未写线程，则随机选择阻塞的读或写线程，进行执行

写者优先：无论当前为读、写线程运行，优先选择阻塞的写线程

                 只有当无阻塞的写线程时，阻塞的读线程才获取执行机会

公平竞争：读写线程按照先来后到的顺序(FIFO)，依次执行，需要用到队列数据结构

 

 java来解决该问题有很多常用的方法，包括使用 wait()与notify() ，synchronized，使用Semaphore信号量，以及jdk1.5+的其他并发技术，下面以读者优先为例来说明。

 

1、使用读锁与写锁来解决，这种方法最简单直观，性能也比较好

 

package com.xx.concurrent.commonUse;

import java.util.concurrent.CountDownLatch;

public class ReaderAndWriterWithMonitor {
	
	//读锁
	static Object w = new Object();
	//写锁
	static Object r = new Object();
	
	//内容 
	static int data = 0 ;

	static CountDownLatch latch = new CountDownLatch(150);

	class Reader extends Thread {

		int quantity;

		Reader(int quantity) {
			this.quantity = quantity;
		}

		@Override
		public void run() {
			synchronized (w) {
					while (quantity > 0) {
						System.out.println(getName()
								+ " is reading ...【data=" + data + "】");
						quantity--;
					}
					latch.countDown();
			}
		}
	}

	class Writer extends Thread {
		int quantity;

		Writer(int quantity) {
			this.quantity = quantity;
		}

		@Override
		public void run() {
			synchronized (w) {
				synchronized (r) {
					while (quantity > 0) {
						data++;
						System.out.println(getName() + " is writing...【data=" + data + "】");
						quantity--;
					}
					latch.countDown();
				}
			}
		}
	}

	public static void main(String[] args) throws InterruptedException {
		long startTime = System.nanoTime();
		ReaderAndWriterWithMonitor demo = new ReaderAndWriterWithMonitor();
		for (int i = 0; i < 100; ++i) {
			demo.new Reader(10).start();
		}
		for (int i = 0; i < 50; ++i) {
			demo.new Writer(1).start();
		}

		latch.await();
		long endTime = System.nanoTime();
		System.out.println(endTime - startTime + "ns");
	}

}

 2、使用信号量

package com.xx.concurrent.commonUse;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;

public class ReaderAndWriter {
	
	static Semaphore mutex = new Semaphore(1);
	static Semaphore w = new Semaphore(1);
	static int readcnt = 0 ;
	static CountDownLatch latch = new CountDownLatch(150);
	static int data = 0;
	
	class Reader extends Thread{
		
		int quantity;
		
		Reader(int quantity){
			this.quantity = quantity;
		}
		
		@Override
		public void run(){
			while(quantity > 0){
				try {
					mutex.acquire();
					readcnt++;
					if (readcnt == 1)
						w.acquire();
					mutex.release();
					//read something
					System.out.println(getName() + " is reading ...【data=" + data + "】");
					mutex.acquire();
					readcnt--;
					if (readcnt == 0)
						w.release();
					mutex.release();
					quantity--;
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
			latch.countDown();
		}
	}
	
	class Writer extends Thread{
		int quantity;
		
		Writer(int quantity){
			this.quantity = quantity;
		}
		
		@Override
		public void run(){
			while(quantity>0){
				try {
					w.acquire();
					data++;
					System.out.println(getName() + " is writing ...【data=" + data + "】");
					w.release();
					quantity--;
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
			latch.countDown();
		}
	}
	
	
	public static void main(String[] args) throws InterruptedException {
		long startTime = System.nanoTime();
		ReaderAndWriter demo = new ReaderAndWriter();
		ExecutorService service = Executors.newFixedThreadPool(150);
		for(int i=0; i< 100; ++i){
			 service.execute(demo.new Reader(10));
		}
		for(int i=0 ; i< 50; ++i){
			service.execute(demo.new Writer(1));
		}
		latch.await();
		service.shutdown();
		long endTime = System.nanoTime();
		System.out.println(endTime - startTime + "ns");
		
	}
}

 

 3、使用wait和notify方式

package com.xx.concurrent.commonUse;

import java.util.concurrent.CountDownLatch;

public class ReaderAndWriterWithWaitNotify {

	static Object w = new Object();

	static int readcnt = 0;
	static int writecnt = 0;
	
	static int data = 0;

	static CountDownLatch latch = new CountDownLatch(150);

	class Reader extends Thread {

		int quantity;

		Reader(int quantity) {
			this.quantity = quantity;
		}

		@Override
		public void run() {
			synchronized (w) {
				try {
					while (writecnt > 0) {
						w.wait();
					}
					readcnt++;
					while (quantity > 0) {
						System.out.println(getName() + " is reading...【data=" + data + "】" );
						quantity--;
					}
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
				readcnt--;
				w.notify();
				latch.countDown();
			}
		}
	}

	class Writer extends Thread {
		int quantity;

		Writer(int quantity) {
			this.quantity = quantity;
		}

		@Override
		public void run() {
			synchronized (w) {
				while (readcnt > 0 || writecnt > 0) {
					try {
						w.wait();
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
				}
				writecnt++;
				while (quantity > 0) {
					data++;
					System.out.println(getName() + " is writing...【data=" + data + "】" );
					quantity--;
				}
				writecnt--;
				w.notify();
				latch.countDown();
			}
		}
	}

	public static void main(String[] args) throws InterruptedException {
		long startTime = System.nanoTime();
		ReaderAndWriterWithWaitNotify demo = new ReaderAndWriterWithWaitNotify();
		for (int i = 0; i < 100; ++i) {
			demo.new Reader(10).start();
		}
		for (int i = 0; i < 50; ++i) {
			demo.new Writer(1).start();
		}

		latch.await();
		long endTime = System.nanoTime();
		System.out.println(endTime - startTime + "ns");

	}

}