生产者消费者---简易消息队列实现

超、凢脫俗 2023-07-24 13:21 24阅读 0赞

**目录**

1，场景

2，源码

a）消息体

b）消息队列

c）生产者

d）消费者

e）测试用例

f）结果输出

--------------------

# 1，场景 #

适用场景太多，不做描述。直接上源码

# 2，源码 #

## a）消息体 ##

package com.wayne.producerConsumer;
    
    public class Message {
        
        private String data;
        
        public Message(String data){
            this.data = data;
        }
    
        public String getData() {
            return data;
        }
    }

## b）消息队列 ##

package com.wayne.producerConsumer;
    
    import java.util.LinkedList;
    
    public class MessageQueue {
    
        private final LinkedList<Message> queue;
    
        private final static int DEFAULT_QUEUE_SIZE = 100;
    
        private final int limit;
    
        public MessageQueue(){
            this(DEFAULT_QUEUE_SIZE);
        }
    
        public MessageQueue(int limit){
            this.limit = limit;
            this.queue = new LinkedList<>();
        }
    
        /**
         * 放消息
         */
        public void put(final Message message) throws InterruptedException {
            synchronized (queue){
                if(queue.size()>=limit){
                    queue.wait();
                }
    
                queue.addLast(message);
                queue.notifyAll();
            }
        }
    
        /**
         * 拿消息
         */
        public Message get() throws InterruptedException {
            synchronized (queue){
                if(queue.isEmpty()){
                    queue.wait();
                }
    
                Message message = queue.removeFirst();
                queue.notifyAll();
                return message;
            }
        }
    
        /**
         * 获取最大值
         */
        public int getLimit(){
            return limit;
        }
    
        /**
         * 获取当前消息队列的长度
         */
        public int getQueueSize(){
            synchronized (queue){
                return queue.size();
            }
        }
    }

## c）生产者 ##

package com.wayne.producerConsumer;
    
    import java.util.Random;
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.atomic.AtomicInteger;
    
    public class Producer extends Thread {
    
        private final MessageQueue queue;
    
        public final static Random random = new Random(System.currentTimeMillis());
    
        public final static AtomicInteger counter = new AtomicInteger(0);
    
        public Producer(MessageQueue queue,int seq){
            super("Producer-"+seq);
            this.queue = queue;
        }
    
        @Override
        public void run() {
            while (true){
                try {
                    Message message = new Message("Message-" + counter.getAndIncrement());
                    this.queue.put(message);
                    TimeUnit.MILLISECONDS.sleep(random.nextInt(1000));
                    System.out.println(currentThread().getName()+"put message:"+message.getData());
                } catch (InterruptedException e) {
                    break;
                }
            }
        }
    }

## d）消费者 ##

package com.wayne.producerConsumer;
    
    import java.util.Random;
    import java.util.concurrent.TimeUnit;
    
    public class Consumer extends Thread {
        private final MessageQueue queue;
    
        private final Random random = new Random(System.currentTimeMillis());
    
        public Consumer(MessageQueue queue,int seq){
            super("Consumer-"+seq);
            this.queue = queue;
        }
    
        @Override
        public void run() { 
            while (true){
                try {
                    Message message = this.queue.get();
                    TimeUnit.MILLISECONDS.sleep(random.nextInt(1000));
                    System.out.println(currentThread().getName()+"get message:"+message.getData());
                } catch (InterruptedException e) {
                    break;
                }
            }
        }
    }

## e）测试用例 ##

package com.wayne.producerConsumer;
    
    import java.util.stream.IntStream;
    
    public class ProducerConsumer {
    
        public static void main(String[] args) {
            MessageQueue messageQueue = new MessageQueue();
    
            new Producer(messageQueue,1).start();
            new Producer(messageQueue,2).start();
            new Producer(messageQueue,3).start();
    
            new Consumer(messageQueue,1).start();
            new Consumer(messageQueue,2).start();
        }
    }

## f）结果输出 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2E4MEM1MQ_size_16_color_FFFFFF_t_70][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2E4MEM1MQ_size_16_color_FFFFFF_t_70]: /images/20230528/715f4290ad71416abb5fe7725225d943.png