1.消费者生产者
import java.util.LinkedList;
import java.util.Queue;
/*
* 生产者的主要作用是生成一定量的数据放到缓冲区中，然后重复此过程。
* 与此同时，消费者也在缓冲区消耗这些数据。
* 该问题的关键就是要保证生产者不会在缓冲区满时加入数据，消费者也不会在缓冲区中空时消耗数据。
*
*
* 简单来说就是生产者不断的生产资源，消费者不断生产资源，
* 但是生产者不会在缓冲区满时加入数据，消费者也不会在缓冲区中空时消耗数据
*
*
* 互斥关系：产者和消费者是同步运行的，肯定会遇到两者同时用缓存区的时候，
* 如果同时用的话，那么产品数，和消费数就会不一致，所以两者是互斥关系，对于这样的问题就必须对缓冲区进行互斥。
*
*
*
*/
public class 消费者生产者 {
public static void main(String[] args) {
Buffer buffer = new Buffer();
Consumer consumer = new Consumer(buffer);
Producer producer = new Producer(buffer);
producer.start();
consumer.start();
}
}
class Producer extends Thread{
private Buffer buffer;
public Producer(Buffer buffer) {
this.buffer = buffer;
}
public void run() {
for(int i=0;i<10;i++) {
try {
buffer.add(i);
Thread.sleep(500);//模拟生产者需要生产时间
System.out.println("生产"+i);
} catch (InterruptedException e1) {
e1.printStackTrace();
}
}
}
}
class Consumer extends Thread{
private Buffer buffer;
public Consumer(Buffer buffer) {
this.buffer = buffer;
}
@Override
public void run() {
for(int i=0;i<10;i++) {
try {
int val =buffer.pull(); // 消费者不断消费就可以
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("消费"+i);
}
}
}
//这里创建一个有限的缓冲区，即临界资源，用于实现生产者和消费者的互斥
//synchronized是Java中的关键字，是一种同步锁
class Buffer{
private Queue<Integer> queue = new LinkedList<>();
// 创建一个队列模拟临界资源
private int size=5;
// 在往生产者里边生产的时候，
//如果缓冲区已经满了，呢么让生产者进程等待
public synchronized void add(int val) throws InterruptedException {
if(queue.size()>size) {
wait();// 不让生产者继续生产，
}
queue.add(val);
notify();// 通知消费者去消费
}
//在消费者消费资源的时候，
//如果缓冲区是空的，呢么让消费者进程等待
public synchronized int pull() throws InterruptedException {
if(queue.size()==0) {
wait();// 没有资源可以消费，阻塞消费者进程
}
int val=queue.poll();
notify();
return val;
}
}
2.哲学家进餐问题
package 实验一;
//每个哲学家都拿着左手的筷子，呢么就会造成死锁
//为了解决这个问题只当哲学家发现左右两边的筷子都可用时，才拿起筷子，
//否则等待。哲学家编号及筷子编号，哲学家i左手边的筷子编号为(i+1)%5，右手边的筷子编号为i。
public class 哲学家进餐问题 {
public static void main(String[] args) {
//创建五个哲学家线程
Philosopher p1=new Philosopher(0);
Philosopher p2=new Philosopher(1);
Philosopher p3=new Philosopher(2);
Philosopher p4=new Philosopher(3);
Philosopher p5=new Philosopher(4);
/*
* public Thread(Runnable target, String name) {
this(null, target, name, 0);
}
*/
new Thread(p1,"0").start();
new Thread(p2,"1").start();
new Thread(p3,"2").start();
new Thread(p4,"3").start();
new Thread(p5,"4").start();
}
}
class Philosopher extends Thread{
private int index;
public static Chopstick chop=new Chopstick();
public Philosopher(int index) {
this.index = index;
}
public synchronized void thinking(){
System.out.println("哲学家"+index+"在思考");
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
public synchronized void eating(){
System.out.println("哲学家"+index+"在吃饭");
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
@Override
public void run() {
while(true){
//思考之后，去拿筷子吃饭
thinking();
chop.takeChopsticks(index);
//吃完之后，去思考
eating();
chop.putChopsticks(index);
}
}
}
// 哲学家拿筷子和放筷子是互斥的
class Chopstick {
public boolean[] isUsing=new boolean[5];// 用来记录筷子的占用情况
public synchronized void takeChopsticks(int index){
while(isUsing[index]||isUsing[(index+1)%5]){//当左右手有一个筷子是已经被占用的时候，就要阻塞这个哲学家进程
try {
wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
//进行到这，说明这个哲学家左右手的筷子都是空闲的
isUsing[index]=true;
isUsing[(index+1)%5]=true;
System.out.println("哲学家"+index+"拿起筷子");
}
public synchronized void putChopsticks(int index){
isUsing[index]=false;
isUsing[(index+1)%5]=false;
System.out.println("哲学家"+index+"放下筷子");
notify();// 放下筷子吃完饭 唤醒这个哲学家进程去拿筷子吃饭
}
}
3.读者写者问题
package 实验一;
import java.util.concurrent.Semaphore;
/*
* 写进程与写进程之间必须互斥的写入数据
* 写进程与读进程之间必须互斥的访问共享数据
* 读进程与读进程之间可以同时访问数据，不需要实现互斥的访问共享数据
*
*
* 写进程与写进程之间必须互斥的写入数据
* 写进程与读进程之间必须互斥的访问共享数据
* 读进程与读进程之间可以同时访问数据，不需要实现互斥的访问共享数据
*/
/*
* Semaphore（信号量）是用来控制同时访问特定资源的线程数量，通过协调各个线程以保证合理地使用公共资源。
* void acquire() ：从信号量获取一个许可，如果无可用许可前将一直阻塞等待
*
* 其构造函数：
* public Semaphore(int permits) {
sync = new NonfairSync(permits);
}
permits 初始许可数，也就是最大访问线程数
int availablePermits()： 获取当前信号量可用的许可
void release()：释放一个许可
*/
public class 读者写者问题 {
public static void main(String[] args) throws InterruptedException {
int writerNumber=3;
int readerNumber=7;
Disk disks=new Disk();
Thread[] writer_threads=new Thread[writerNumber];
Writer[] writers=new Writer[writerNumber];
Thread[] reader_threads=new Thread[readerNumber];
Reader[] readers=new Reader[readerNumber];
/*启动部分写者*/
for (int i=0;i<writerNumber/2;i++)
{
writers[i]=new Writer(disks);
writer_threads[i]=new Thread(writers[i],Integer.toString(i));
writer_threads[i].start();
}
/*暂停等待写者启动*/
Thread.sleep(100);
/*启动读者*/
for (int i=0;i<readerNumber;i++)
{
readers[i]=new Reader(disks);
reader_threads[i]=new Thread(readers[i],Integer.toString(i));
reader_threads[i].start();
}
for (int i=writerNumber/2;i<writerNumber;i++)
{
writers[i]=new Writer(disks);
writer_threads[i]=new Thread(writers[i],Integer.toString(i));
writer_threads[i].start();
}
}
}
/**
* 模拟被读和被写的对象
*/
class Disk extends Thread{
private String data_str; /*读写数据*/
private int reader_count; /*读者数量*/
private Semaphore write_mutex; /*写者信号量*/
private Semaphore read_mutex; /*读者信号量*/
private Semaphore read_count_mutex; /*修改读者计数的信号量 用于修改reader_count时使用的pv操作*/
Disk()
{
write_mutex=new Semaphore(1);
read_mutex=new Semaphore(1000);
read_count_mutex=new Semaphore(1);
}
public void start_read() throws InterruptedException {
//有写者时忙等
while (write_mutex.availablePermits()==0){ }//int availablePermits()： 获取当前信号量可用的许可
// 修改读者计数pv操作
read_count_mutex.acquire();//申请
reader_count++;
read_count_mutex.release();//释放
/*获取信号量*/
read_mutex.acquire();
}
public void finish_read() throws InterruptedException {
/*修改读者计数 （pv操作）*/
read_count_mutex.acquire();
reader_count--;
read_count_mutex.release();
read_mutex.release();//释放读者的信号量
}
public void start_write() throws InterruptedException {
/*获取写者信号量*/
write_mutex.acquire();
/*当有读者时忙等*/
while (getReader_count()!=0){}
}
public void finish_write()
{
/*释放写者信号量*/
write_mutex.release();
}
public String read() throws InterruptedException {
return data_str;
}
public String write(String new_data) throws InterruptedException {
this.data_str=new_data;
return data_str;
}
public int getReader_count() {
return reader_count;
}
public void setReader_count(int reader_count) {
this.reader_count = reader_count;
}
}
class Reader extends Thread {
private Disk disk;
public Reader(Disk disk) {
this.disk = disk;
}
@Override
public void run() {
for (int i=10;i>=0;i--)
{
/*读操作*/
try {
read();
} catch (InterruptedException e) {
e.printStackTrace();
}
//模拟真实的读操作
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
private void read() throws InterruptedException {
disk.start_read();
/*读数据并打印*/
String str=disk.read();
System.out.println("读操作："+Thread.currentThread().getId()+" 现在的数据为："+str);
disk.finish_read();
}
}
class Writer extends Thread{
private Disk disk;
public Writer(Disk disk) {
this.disk = disk;
}
public void run() {
for (int i=10;i>=0;i--)
{
try {
write();
} catch (InterruptedException e) {
e.printStackTrace();
}
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
private void write() throws InterruptedException {
disk.start_write();
String str=disk.write("data+"+Thread.currentThread().getId());
System.out.println("写操作："+Thread.currentThread().getId()+" 现在的数据为："+str);
disk.finish_write();// 释放写者信号量,让别的写者可以接着写入
}
}
