生产者使用者使用线程 [英] Producer Consumer using threads

查看:121
本文介绍了生产者使用者使用线程的处理方法,对大家解决问题具有一定的参考价值,需要的朋友们下面随着小编来一起学习吧!

问题描述

我正在编写一个程序,该程序使用多线程概念来实现Java中的Producer Consumer问题.以下是一些我应该怎么做的详细信息:

I’m writing a program that implements the Producer Consumer problem in Java using multithreading concepts. Below are few details how I’m supposed to do it:

1)主线程应创建一个缓冲区,其容量指定为命令行参数.生产者线程和使用者线程的数量也指定为命令行参数.我应该为每个生产者和使用者线程分配一个唯一的编号.如何为生产者线程和使用者线程分配唯一编号?

1) The main thread should create a buffer with capacity specified as a command line argument. The number of producer and consumer threads are also specified as command line arguments. I’m supposed to assign a unique number to each producer and consumer thread. How do I assign a unique number to producer and consumer threads?

2)生产者线程在无限循环中运行.它产生具有以下格式的数据项(字符串):<producer number>_<data item number>.例如,线程1的第一个数据项将是1_1,线程3的第二个数据项将是3_2.如何以这种格式创建数据项?

2) The producer thread operates in an infinite loop. It produces a data item (a string) with the following format: <producer number>_<data item number>. For example the 1st data item from thread number 1 will be 1_1 and second data item from thread number 3 will be 3_2. How do create data items in such a format?

3)然后,生产者线程将一个条目写入生产者日志文件(<生产者编号>"Generated" <data item>).写入日志条目后,它将尝试插入缓冲区.如果插入成功,它将在日志文件中创建一个条目(<producer number> <data item>插入成功").我该如何编写这样的代码?

3) Then the Producer thread writes an entry into the producer log file (< producer number > "Generated" <data item>). Upon writing the log entry, it attempts to insert into the buffer. If insertion is successful, it creates an entry into the log file (<producer number> <data item> "Insertion successful"). How do I write such a code?

下面是我编写的Java代码.

Below is the Java code I wrote.

import java.util.*;
import java.util.logging.*;

public class PC2
{
    public static void main(String args[])
    {
            ArrayList<Integer> queue = new ArrayList<Integer>();

            int size = Integer.parseInt(args[2]);
            Thread[] prod = new Thread[Integer.parseInt(args[0])];
            Thread[] cons = new Thread[Integer.parseInt(args[1])];

            for(int i=0; i<prod.length; i++)
            {
                    prod[i] = new Thread(new Producer(queue, size));
                    prod[i].start();
            }

            for(int i=0; i<cons.length; i++)
            {
                    cons[i] = new Thread(new Consumer(queue, size));
                    cons[i].start();
                }

    }
}

class Producer extends Thread
{
    private final ArrayList<Integer> queue;
    private final int size;

    public Producer(ArrayList<Integer> queue, int size)
    {
            this.queue = queue;
            this.size = size;
    }

    public void run()
    {
            while(true){
            for(int i=0; i<size; i++)
            {
                    System.out.println("Produced: "+i+" by id " +Thread.currentThread().getId());
try
                    {
                            produce(i);
                            Thread.sleep(3000);
                    }
                    catch(Exception e)
                    {
                            Logger.getLogger(Producer.class.getName()).log(Level.SEVERE, null, e);
                    }
            }}
    }


    public void produce(int i) throws InterruptedException
    {
            while(queue.size() == size)
            {
                    synchronized(queue)
                    {
                            System.out.println("Queue is full "+Thread.currentThread().getName() +" is waiting, size: "+queue.size());
                            queue.wait();
                       }
            }
            synchronized(queue)
            {
                    queue.add(i);
                    queue.notifyAll();
            }
    }
}

class Consumer extends Thread
{
    private final ArrayList<Integer> queue;
    private final int size;

    public Consumer(ArrayList<Integer> queue, int size)
    {
            this.queue = queue;
            this.size = size;
    }

    public void run()
    {
            while(true)
            {
                    try
                    {       System.out.println("Consumed: "+consume());
                            Thread.sleep(1000);
                    }
                    catch(Exception e)
                    {
                            Logger.getLogger(Consumer.class.getName()).log(Level.SEVERE, null, e);
                    }
            }
    }

    public int consume() throws InterruptedException
    {
            while(queue.isEmpty())
            {
                    synchronized(queue)
                    {
                            System.out.println("Queue is empty "+Thread.currentThread().getName()+" is waiting, size: "+queue.size());
                            queue.wait();
                        }
            }

            synchronized (queue)
            {
                    queue.notifyAll();
                    System.out.println("Consumed by id "+Thread.currentThread().getId());
                    return (Integer) queue.remove(0);

            }
    }
}

如何执行上述步骤?

推荐答案

对于生产者消费者问题,最佳解决方案是BlockingQueue.我正在测试一些东西,因此设计相同的程序,现在可以根据您的需要对其进行修改.

For producer consumer problem best solution is BlockingQueue. I was testing a few things so designed same kind of program now modified it as per your need.

看看是否有帮助.

import java.util.concurrent.*;
public class ThreadingExample {

    public static void main(String args[]){
        BlockingQueue<Message> blockingQueue = new ArrayBlockingQueue<Message>(100);
        ExecutorService exec = Executors.newCachedThreadPool();
        exec.execute(new Producer(blockingQueue));
        exec.execute(new Consumer(blockingQueue));
    }

}
class Message{
    private static int count=0;
    int messageId;
    Message(){
        this.messageId=count++;
        System.out.print("message Id"+messageId+" Created ");
    }
}
class Producer implements Runnable{

    private BlockingQueue<Message> blockingQueue;
    Producer(BlockingQueue<Message> blockingQueue){
        this.blockingQueue=blockingQueue;
    }

    @Override
    public void run(){
        while(!Thread.interrupted()){
            System.out.print("Producer Started");
            try {
                blockingQueue.put(new Message());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("Producer Done");
        }
    }
}

class Consumer implements Runnable{
    private BlockingQueue<Message> blockingQueue;
    Consumer(BlockingQueue<Message> blockingQueue){
        this.blockingQueue=blockingQueue;
    }

    @Override
    public void run(){
        while(!Thread.interrupted()){
            System.out.print("Concumer Started");
            try{
                Message message  = blockingQueue.take();
                System.out.print("message Id"+message.messageId+" Consumed ");
            }
            catch(InterruptedException e){
                e.printStackTrace();
            }
            System.out.println("Concumer Done");
        }
    }
}

这篇关于生产者使用者使用线程的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持IT屋!

查看全文
登录 关闭
扫码关注1秒登录
发送“验证码”获取 | 15天全站免登陆