多线性打印数字

多线性的面试题目不在少数。可是题目似乎不多。找了找，发现这个题目：

需要用多线性来实现：

您好！
本帖隐藏的内容需要积分高于 200 才可浏览
您当前积分为 0。
使用VIP即刻解锁阅读权限或查看其他获取积分的方式

游客，您好！
本帖隐藏的内容需要积分高于 200 才可浏览
您当前积分为 0。
VIP即刻解锁阅读权限 或 查看其他获取积分的方式
Unlock interview details and practice with AI
Curated Interview Questions from Top Companies

有什么更好的方法，或者局部修正，请不吝赐教。

每个工人通知下一个工人的想法是很好的，不过代码里把同步机制（各种lock、barrier，还有synchronized关键字）与真正的业务逻辑（也就是累加并打印数字）混合在一起了，看起来会比较乱

所以一个改进是如何把同步机制给解耦（decoupling）出来，这里常用的一种抽象是BlockingQueue：各个工人之间通过阻塞队列发送接收消息，而所有同步操作都可以封装在阻塞队列里面

1. import java.util.ArrayList;
   
2. import java.util.LinkedList;
   
3. import java.util.Queue;
   
4. import java.util.concurrent.locks.Condition;
   
5. import java.util.concurrent.locks.Lock;
   
6. import java.util.concurrent.locks.ReentrantLock;
   
7. 
   
8. public class Main {
   
9.   public static void main(String[] args) throws InterruptedException {
   
10.     // 参数
    
11.     final int K = 10;
    
12.     final int N = 3;   
    
13.     final int MAX_VALUE = K * N;
    
14. 
    
15.     // 初始化N个消息管道，这份代码里手写了一个简易的阻塞消息管道
    
16.     // 也可以用Java自带的LinkedBlockingDeque
    
17.     ArrayList<SimpleBlockingQueue<Integer>> messageQueues = new ArrayList<>();
    
18.     for (int i = 0; i < N; ++i) {
    
19.       messageQueues.add(new SimpleBlockingQueue<Integer>());
    
20.     }
    
21.    
    
22.     // 初始化N个worker，每一个worker对应一个线程
    
23.     ArrayList<Thread> workerThreads = new ArrayList<>();
    
24.     for (int i = 0; i < N; ++i) {
    
25.       // 当前worker的outgoing消息队列即是下一个worker的incoming消息队列，构成一个环路
    
26.       Worker worker = new Worker(i, MAX_VALUE,
    
27.                                  messageQueues.get(i),
    
28.                                  messageQueues.get(i == N - 1 ? 0 : i + 1));
    
29.       Thread thread = new Thread(worker);
    
30.       thread.start();
    
31.       workerThreads.add(thread);
    
32.     }
    
33. 
    
34.     // 给0号worker发送启动消息
    
35.     messageQueues.get(0).add(0);
    
36.    
    
37.     // 等待所有线程执行结束
    
38.     for (Thread thread : workerThreads) {
    
39.       thread.join();
    
40.     }
    
41.   }
    
42. }
    
43. 
    
44. class Worker implements Runnable {
    
45.   private int workerId;
    
46.   private int maxValueExclusive;
    
47.   private SimpleBlockingQueue<Integer> incomingMessages;
    
48.   private SimpleBlockingQueue<Integer> outgoingMessages;
    
49. 
    
50.   public Worker(int workerId, int maxValueExclusive,
    
51.                 SimpleBlockingQueue<Integer> incomingMessages,
    
52.                 SimpleBlockingQueue<Integer> outgoingMessages) {
    
53.     this.workerId = workerId;
    
54.     this.maxValueExclusive = maxValueExclusive;
    
55.     this.incomingMessages = incomingMessages;
    
56.     this.outgoingMessages = outgoingMessages;
    
57.   }
    
58.   
    
59.   @Override
    
60.   public void run() {
    
61.     while (true) {
    
62.       Integer token = null;
    
63.       // 从上一个worker那里获得token，当前持有token的worker才被允许打印数字
    
64.       // 为了简洁，token本身就是一个Integer顺便记录了需要打印的数字
    
65.       try {
    
66.         token = incomingMessages.take();
    
67.       } catch (InterruptedException e) {}
    
68.       // 出错或者是超过最大值的情况
    
69.       if (token == null || token >= maxValueExclusive) {
    
70.         // 仍然需要传递token来通知下游worker终止运行
    
71.         outgoingMessages.add(maxValueExclusive);
    
72.         break;
    
73.       }
    
74.       System.out.println("Thread " + workerId + " is now printing " + token);
    
75.       outgoingMessages.add(token + 1);      
    
76.     }
    
77.   }
    
78. }
    
79. 
    
80. // 简易的BlockingQueue实现
    
81. class SimpleBlockingQueue<E> {
    
82.   private Lock lock = new ReentrantLock();
    
83.   private Condition isEmptyCond = lock.newCondition();
    
84.   private Queue<E> items = new LinkedList<>();
    
85.   
    
86.   public void add(E item) {
    
87.     lock.lock();
    
88.     items.add(item);
    
89.     if (items.size() == 1) {
    
90.       isEmptyCond.signalAll();
    
91.     }
    
92.     lock.unlock();
    
93.   }
    
94.   
    
95.   public E take() throws InterruptedException {
    
96.     lock.lock();
    
97.     try {
    
98.       while (items.isEmpty()) {
    
99.         isEmptyCond.await();
    
100.       }
     
101.       return items.poll();
     
102.     } finally {
     
103.       lock.unlock();
     
104.     }
     
105.   }
     
106. }

复制代码

另一种更具一般性的实现方式是引入一个协调（调度）线程，也就是常说的coordinator

然后worker和coordinator之间通过消息通信，各自有一个event loop根据消息类型执行对应的操作 —— 这种并发模型称作actor model：https://en.wikipedia.org/wiki/Actor_model

1. import java.util.ArrayList;
   
2. import java.util.concurrent.BlockingQueue;
   
3. import java.util.concurrent.LinkedBlockingQueue;
   
4. 
   
5. public class Main {
   
6.   public static void main(String[] args) throws InterruptedException {
   
7.     // 参数
   
8.     final int K = 10;
   
9.     final int N = 3;   
   
10.     final int MAX_VALUE = K * N;
    
11. 
    
12.     // 创建并启动coordinator线程
    
13.     Thread coordinatorThread = new Thread(new Coordinator(N, MAX_VALUE));
    
14.     coordinatorThread.start();
    
15.     coordinatorThread.join();
    
16.   }
    
17. }
    
18. 
    
19. class Coordinator implements Runnable {
    
20.   private ArrayList<Worker> workers = new ArrayList<>();
    
21.   private ArrayList<Thread> workerThreads = new ArrayList<>();
    
22.   private BlockingQueue<Message> incomingMessages = new LinkedBlockingQueue<>();
    
23.   private int maxValue;
    
24.   
    
25.   public Coordinator(int numWorkers, int maxValue) {
    
26.     // Coordinator负责创建并管理所有worker
    
27.     for (int i = 0; i < numWorkers; ++i) {
    
28.       Worker worker = new Worker(i, this);
    
29.       workers.add(worker);
    
30.       workerThreads.add(new Thread(worker));
    
31.     }
    
32.     this.maxValue = maxValue;
    
33.   }
    
34. 
    
35.   public void sendMessage(Message message) {
    
36.     incomingMessages.add(message);
    
37.   }
    
38. 
    
39.   @Override
    
40.   public void run() {
    
41.     // 启动所有worker线程
    
42.     for (Thread thread : workerThreads) {
    
43.       thread.start();
    
44.     }
    
45.     // 依次向各个worker发送打印数字的消息
    
46.     for (int i = 0; i < maxValue; ++i) {
    
47.       Worker worker = workers.get(i % workerThreads.size());
    
48.       worker.sendMessage(new PrintValueMessage(i));
    
49.       // 等待收到worker回应后才开始下一个循环
    
50.       try {
    
51.         incomingMessages.take();
    
52.       } catch (InterruptedException e) {}
    
53.     }
    
54.     // 完成所有打印后，向各个worker发送终止消息
    
55.     for (Worker worker : workers) {
    
56.       worker.sendMessage(new JoinMessage());
    
57.     }
    
58.     // 等待worker线程终止
    
59.     for (Thread thread : workerThreads) {
    
60.       try {
    
61.         thread.join();
    
62.       } catch (InterruptedException e) {}
    
63.     }   
    
64.   }
    
65. }
    
66. 
    
67. class Worker implements Runnable {
    
68.   private int workerId;
    
69.   private BlockingQueue<Message> incomingMessages =
    
70.       new LinkedBlockingQueue<>();
    
71.   private Coordinator coordinator;
    
72. 
    
73.   public Worker(int workerId, Coordinator coordinator) {
    
74.     this.workerId = workerId;
    
75.     this.coordinator = coordinator;
    
76.   }
    
77.   
    
78.   public void sendMessage(Message message) {
    
79.     incomingMessages.add(message);
    
80.   }
    
81.    
    
82.   @Override
    
83.   public void run() {
    
84.     boolean running = true;
    
85.     while (running) {
    
86.       // 等待coordinator发来消息
    
87.       Message msg = null;
    
88.       try {
    
89.         msg = incomingMessages.take();
    
90.       } catch (InterruptedException e) {
    
91.         continue;
    
92.       }
    
93.       // 消息有两种：打印数字或者是终止运行，根据消息类型执行对应操作
    
94.       switch (msg.getKind()) {
    
95.         case PRINT_VALUE:
    
96.           System.out.println("Thread " + workerId + " is now printing " +
    
97.                              ((PrintValueMessage)msg).getValue());
    
98.           coordinator.sendMessage(new AckMessage());
    
99.           break;
    
100.         case JOIN:
     
101.           running = false;
     
102.           break;
     
103.         default:
     
104.           System.err.println("Unexpected message received");
     
105.       }
     
106.     }
     
107.   }
     
108. }
     
109. 
     
110. // 这里添加了三种消息类型，还可以扩展添加各种各样的消息
     
111. enum MessageKind {
     
112.   PRINT_VALUE,
     
113.   JOIN,
     
114.   ACK
     
115. }
     
116. 
     
117. abstract class Message {
     
118.   private MessageKind kind;
     
119. 
     
120.   public Message(MessageKind kind) {
     
121.     this.kind = kind;
     
122.   }
     
123.   public MessageKind getKind() {
     
124.     return kind;
     
125.   }  
     
126. }
     
127. 
     
128. class PrintValueMessage extends Message {
     
129.   private final int value;
     
130. 
     
131.   public PrintValueMessage(int value) {
     
132.     super(MessageKind.PRINT_VALUE);
     
133.     this.value = value;
     
134.   }  
     
135.   public int getValue() {
     
136.     return value;
     
137.   }  
     
138. }
     
139. 
     
140. class JoinMessage extends Message {
     
141.   public JoinMessage() {
     
142.     super(MessageKind.JOIN);
     
143.   }
     
144. }
     
145. 
     
146. class AckMessage extends Message {
     
147.   public AckMessage() {
     
148.     super(MessageKind.ACK);
     
149.   }
     
150. }
     

复制代码

问下你怎么保证line48能获取到了锁？
你提到notifyall这个效率问题，用condition来隔离呼唤单个线程应该就行了吧。

yuqinlear 发表于 2019-6-1 12:07
问下你怎么保证line48能获取到了锁？
你提到notifyall这个效率问题，用condition来隔离呼唤单个线程应该 ...

在每个进程的一开始都有synchronized (myLock)，然后barrier保证了每个工人都拿到了锁。如果48行的锁owner工人没有放弃monitor，那么本工人是无法在48行拿到锁的，会一直在那里等待owner工人进入wait。

对。condition也很方便。基本上是一样的操作。需要await和lock，unlock配合使用。另外每个线程需要一个条件来判断目前该await还是被notify了。

刚刚发现leetcode现在也有了多线性的问题了。