目录
producer 是生产者的意思： 指生产数据的线程，
consumer 是消费者的意思： 指的是使用数据的线程
?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
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(message);
System.out.println(Thread.currentThread().getName() + " put message " + message.getData());
Thread.sleep(random.nextInt(1000));
} catch (InterruptedException e) {
break;
}
}
}
}
?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
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;
}
}
}
}
?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
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;
}
}
?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
public class MessageQueue {
private final static int DEFAULT_MAX_LIMIT = 100;
private final LinkedList<Message> 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();
}
}
}
?
1
2
3
4
5
6
7
8
9
10
11
12
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
到此这篇关于Java多线程 Producer and Consumer设计模式的文章就介绍到这了,更多相关Java多线程 Producer、Consumer内容请搜索服务器 之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持服务器之家！
原文链接：https://juejin.cn/post/7023247132865855524
如果您对该产品感兴趣，请填写办理（客服微信：xiaoxiongyidong）
如果您对该产品感兴趣，请填写办理（客服微信：xiaoxiongyidong）
发表评论
◎欢迎参与讨论，请在这里发表您的看法、交流您的观点。