Java多线程 Producer and Consumer设计模式

Java多线程 Producer and Consumer设计模式

目录

producer是生产http://者的意思：指生产数据的线程，

consumer是消费者的意思：指的是使用数据的线程

public class ProducerThread extends Thread {

private final static Random random = new Random(System.currentTimeMillis());

private final static AtomicInteger counter = new AtomicInteger(0);

private final MessageQueue messageQueue;

public ProducerThread(MessageQueue messageQueue, int seq) {

super("Producer-" + seq);

this.messageQueue = messageQueue;

}

@Override

public void run() {

while (true) {

try {

Message message = new Message("Message-" + counter.getAndIncrement());

messageQueue.put(http://message);

System.out.println(Thread.currentThread().getName() + " put message " + message.getData());

Thread.sleep(random.nextInt(1000));

} catch (InterruptedException e) {

break;

}

}

}

}

public class ConsumerThread extends Thread {

private final static Random random = new Random(System.currentTimeMillis());

private final MessageQueue messageQueue;

public ConsumerThread(MessageQueue messageQueue, int seq) {

super("Consumer-" + seq);

this.messageQueue = messageQueue;

}

@Override

public void run() {

while (true) {

try {

Message message = messageQueue.take();

System.out.println(Thread.currentThread().getName() + " take a message " + message.getData());

Thread.sleep(random.nextInt(1000));

} catch (InterruptedException e) {

break;

}

}

}

}

public class Message {

public Message(String data) {

this.data = data;

}

private String data;

public String getData() {

return data;

}

public void setData(String data) {

this.data = data;

}

}

public class MessageQueue {

private final static int DEFAULT_MAX_LIMIT = 100;

private final LinkedList queue;

private final int limit;

public MessageQueue() {

this(DEFAULT_MAX_LIMIT);

}

public MessageQueue(final int limit) {

this.limit = limit;

this.queue = new LinkedList<>();

}

public void put(final Message message) throws InterruptedException {

synchronized (queue) {

while (queue.size() > limit) {

queue.wait();

}

queue.addLast(message);

queue.notifyAll();

}

}

public Message take() throws InterruptedException {

synchronized (queue) {

while (queue.isEmpty()) {

queue.wait();

}

Message message = queue.removeFirst();

queue.notifyAll();

return message;

}

}

public int getMaxLimit() {

return this.limit;

}

public int getMessageSize() {

synchronized (queue) {

return queue.size();

}

}

}

public class ProducerAndConsumerClient {

public static void main(String[] args) {

final MessageQueue messageQueue = new MessageQueue();

new ProducerThread(messageQueue, 1).start();

new ProducerThread(messageQueue, 2).start();

new ProducerThread(messageQueue, 3).start();

new ConsumerThread(messageQueue, 1).start();

new ConsumerThread(messageQueue, 2).start();

}

}

Producer-1 put message Message-0

Producer-3 put message Message-2

Producer-2 put message Message-1

Consumer-1 take a message Message-0

Consumer-2 take a message Message-1

Producer-2 put message Message-3

Consumer-1 take a message Message-2

Producer-2 put message Message-4

Consumer-2 take a message Message-3

Producer-3 put message Message-5

Producer-3 put message Message-6

Producer-3 put message Message-7

Consumer-1 take a message Message-4

Producer-2 put message Message-8

Consumer-2 take a message Message-5

Producer-3 put message Message-9

Producer-1 put message Message-10

Producer-1 put message Message-11

Producer-2 put message Message-12

省略...

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。