阿里筆試-生產者消費者模式

/**
使用“生產者-消費者模式”編寫代碼實現：線程A隨機間隔（10~200ms）按順序生成1到100的數字（共100個），
放到某個隊列中.3個線程B、C、D即時消費這些數據，線程B打印（向控制台）所有被2整除的數，
線程C打印被3整除的數，線程D打印其它數據，要求數字的打印是有序的（從1到100）
限時40分鍾，可以使用IDE及第三方類庫
**/

 先說一下阿里大牛的評價：我下面寫的程序有個問題，如果B線程掛起了，C,D線程運行的話，有可能順序會不一致。當然不能用三個queue，讓B，C,D分別消費，因為太簡單了。

應該這樣，每次取出一個數來，使用wait(),notify()，先讓B運行，然后C運行，然后D運行；就能保證順序了，下面是我寫的程序，暫時沒有時間修改，以后有時間了，會進行修改：

 

 

 

package com.sankuai.qcs.regulation.TestCase;

import lombok.Data;

@Data
public class aliObj implements Comparable<aliObj> {
    private final int intData;

    public boolean BisDone;
    public boolean CisDone;
    public aliObj(int d){
        intData = d;
        BisDone=false;
        CisDone=false;
    }

    public int getData(){
        return intData;
    }
    @Override
    public String toString(){
        return "data:"+intData;
    }


    @Override
    public int compareTo(aliObj o) {

        if(this.intData>o.intData) {
            return 1;
        }
            else if(this.intData<o.intData) {
            return -1;
        }
        else {
            return 0;
        }

    }
}

 

 

package com.sankuai.qcs.regulation.TestCase;

import java.util.Comparator;
import java.util.Random;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.TimeUnit;

public class Producer implements Runnable {

    private volatile boolean isRunning = true;
    int capacity = 100;
    PriorityBlockingQueue<aliObj> queue = new PriorityBlockingQueue(capacity, new Comparator<aliObj>() {
        @Override
        public int compare(aliObj i1, aliObj i2) {
            return i2.getIntData() - i1.getIntData();
        }
    });

    public Producer(PriorityBlockingQueue<aliObj> queue){
        this.queue = queue;
    }

    @Override
    public void run() {
        if (isRunning) {

            Random r = new Random();
            System.out.println("start producting id:" + Thread.currentThread().getId());
            try {

                for (int i = 0; i < 100; i++) {
                    Thread.sleep(10 + r.nextInt(190)); //停頓 10~200毫秒

                    System.out.println(i+1 + " 加入隊列");
                    if (!queue.offer(new aliObj(i + 1), 2, TimeUnit.SECONDS)) {
                        System.err.println(i+1+" 加入隊列失敗");
                    }
                }


            } catch (InterruptedException e) {
                e.printStackTrace();
                Thread.currentThread().interrupt();
            }

        }
    }

    public void stop() {
        isRunning = false;
    }


}

 

 

package com.sankuai.qcs.regulation.TestCase;

import java.util.concurrent.PriorityBlockingQueue;

public class ConsumerB implements Runnable{
    private PriorityBlockingQueue<aliObj> queue;
    public ConsumerB(PriorityBlockingQueue<aliObj> queue){
        this.queue = queue;
    }

    @Override
    public void run() {
        System.out.println("start ConsumerB id :"+Thread.currentThread().getId());

        try{
            while(true){
                aliObj data = queue.take();
                if(data != null)
                {
                    if(data.isBisDone()==false && data.getData()%2==0) {

                        System.out.println("可以被2整除的數："+data);
                        data.setBisDone(true);
                      if(data.getData()%3==0)  queue.offer(data);

                    }

                    else{
                        queue.offer(data);
                    }
                }
            }
        }catch (InterruptedException e) {
            e.printStackTrace();
            Thread.currentThread().interrupt();
        }
    }

}

 

 

package com.sankuai.qcs.regulation.TestCase;

import java.util.concurrent.PriorityBlockingQueue;

public class ConsumerC implements Runnable{
    private PriorityBlockingQueue<aliObj> queue;
    public ConsumerC(PriorityBlockingQueue<aliObj> queue){
        this.queue = queue;
    }

    @Override
    public void run() {
        System.out.println("start ConsumerC id :"+Thread.currentThread().getId());

        try{
            while(true){
                aliObj data = queue.take();
                if(data != null) {
                    if (data.isCisDone()==false &&  data.getData() % 3 == 0) {

                        System.out.println("可以被3整除的數：" + data);

                     if(data.getData()%2==0) {
                         data.setCisDone(true);
                         queue.offer(data);
                     }

                    } else {
                        queue.offer(data);
                    }
                }
                }

        }catch (InterruptedException e) {
            e.printStackTrace();
            Thread.currentThread().interrupt();
        }
    }

}

package com.sankuai.qcs.regulation.TestCase;

import java.util.concurrent.PriorityBlockingQueue;

public class ConsumerD implements Runnable{
    private PriorityBlockingQueue<aliObj> queue;
    public ConsumerD(PriorityBlockingQueue<aliObj> queue){
        this.queue = queue;
    }

    @Override
    public void run() {
        System.out.println("start ConsumerD id :"+Thread.currentThread().getId());

        try{
            while(true){
                aliObj data = queue.take();
                if(data != null)
                {
                    if((data.getData()%2!=0)&&(data.getData()%3!=0)) {

                        System.out.println("既不可以被2整除，也不可以被3整除的數："+data);
                        
                    }
                }
            }
        }catch (InterruptedException e) {
            e.printStackTrace();
            Thread.currentThread().interrupt();
        }
    }

}

package com.sankuai.qcs.regulation.TestCase;
import java.util.concurrent.*;

public class Main {
    public static void main(String[] args) throws InterruptedException {
        PriorityBlockingQueue<aliObj> queue = new PriorityBlockingQueue<>();
        Producer p1 = new Producer(queue);
        ConsumerB b1 = new ConsumerB(queue);

        ConsumerC c1 = new ConsumerC(queue);
        ConsumerD d1 = new ConsumerD(queue);
        ExecutorService service = Executors.newCachedThreadPool();
        service.execute(p1);

        service.execute(b1);
        service.execute(c1);
        service.execute(d1);
        p1.stop();
        Thread.sleep(3000);
        service.shutdown();
    }
}