ZeroMQ(java)之I/O线程的实现与组件间的通讯

算是开始读ZeroMQ（java）的代码实现了吧，如今有了一个大致的了解，看起来实现是比较的干净的，抽象什么的不算复杂。。。

这里先来看看它的I/O线程的实现吧，顺带看看是如何实现组件的通讯的。。。。

首先要搞清楚I/O线程的实现，就先要弄懂一个类型，Poller（zmq.Poller.java），能够将其当作是对selector的一个封装，同时它还要管理定时事件，看了这么多代码，发现基本上都是在实现I/Oselect的地方完成了定时的实现。。。。

好了，不说太多闲话了，来看看它的继承体系吧：

这里还将依赖关系也标出来了，首先继承自PollerBase抽象类，而后实现了Runnable接口，本身还会建立一个Thread对象。。。看了这个图，基本上就已经可以知道Poller的运行原理了吧。。。。

这里先来看看PollerBase的实现吧，它其实主要是用来管理定时的，那么先来看看他的一些重要的属性和定义：

[java]  view plain copy

1. private final AtomicInteger load;   //这个load其实就是当前poller里面注册的channel的数量  
2.   
3. //这里是要注册的超时是事件  
4. private final class TimerInfo {  
5.     IPollEvents sink;  //事件回调  
6.     int id;  
7.       
8.     public TimerInfo(IPollEvents sink_, int id_) {  
9.         sink = sink_;  
10.         id = id_;  
11.     }  
12. }  
13. private final Map<Long, TimerInfo> timers;   //这里记录全部的超时对象，key是时间  
14. private final Map<Long, TimerInfo> addingTimers;   //等待加入的超时事件  

前面的一个原子Integer是用于记录负载的，用于记录当前poller里面一共注册了多少I/O对象。。。而后是超时事件的定义，sink是超时的事件回调函数，里面有相应的方法，timer就记录了全部的超时事件，addingTimers是须要加入的超时事件。。这里的key都是超时的时间，value就是超时对象了。。。

这里就来看两个主要的方法就行了吧，先来看看如何加入超时事件：

[java]  view plain copy

1. //添加一个超时事件  
2. public void add_timer (long timeout_, IPollEvents sink_, int id_) {  
3.     long expiration = Clock.now_ms () + timeout_;   //计算超时的时间  
4.     TimerInfo info = new TimerInfo(sink_, id_);  //建立超时对象  
5.     addingTimers.put(expiration, info);  //将其添加到adding里面去  
6.   
7. }  

代码应该很简单可以看明白吧，第一个参数是超时时间，第二个参数是回调方法，第三个参数是ID，首先加上当前的时间就算出了超时的时间，而后建立超时对象，这里先是将其放入了addingTimers里面，而不是直接放到了timer里面，。。。

那么接下来来看看如何执行全部的超时的方法吧：

[java]  view plain copy

1.  //执行全部的超时事件，返回下一个超时还剩下的时间  
2.     protected long execute_timers() {  
3.         if (!addingTimers.isEmpty()) {  //若是当前还有须要添的超时时间，那么须要将其添加进去  
4.             timers.putAll(addingTimers);  
5.             addingTimers.clear();  
6.         }  
7.         //没有超时事件  
8.         if (timers.isEmpty())  
9.             return 0L;  
10.   
11.         //获取当前的时间  
12.         long current = Clock.now_ms ();  
13.   
14.         //遍历全部的超时时间，这里是从最小的开始的  
15.         Iterator<Entry <Long, TimerInfo>> it = timers.entrySet().iterator();  
16.         while (it.hasNext()) {  
17.   
18.             Entry <Long, TimerInfo> o = it.next();  
19.             //  If we have to wait to execute the item, same will be true about  
20.             //  all the following items (multimap is sorted). Thus we can stop  
21.             //  checking the subsequent timers and return the time to wait for  
22.             //  the next timer (at least 1ms).  
23.   
24.             //若是超时的时间大于当前的时间，那么表示尚未超时，  
25.             if (o.getKey() > current) {  
26.                 return o.getKey() - current;  //返回下一个超时还剩下的时间  
27.             }  
28.   
29.             //  Trigger the timer.  
30.             //执行超时方法  
31.             o.getValue().sink.timer_event (o.getValue().id);  
32.             //  Remove it from the list of active timers.  
33.             it.remove();  
34.         }  
35.   
36.         if (!addingTimers.isEmpty())  
37.             return execute_timers();  
38.   
39.         //  There are no more timers.  
40.   
41.         return 0L;  //若是是0 的话，表示没有timer执行了  
42.     }  
43. }  

应该代码也还算比较好理解吧，这里能够看到将addingTimers里面的都放到了timers里面。。。而后遍历全部的超时对象，并执行他们的超时回调，知道一个超时时间尚未到，最后返回的是下一个超时事件还剩下多长的时间。。。

好了，那么接下来来看看Poller类型的实现吧，先来看看它的重要定义：

[java]  view plain copy

1. //在当前poller里面注册的封装。。。  
2.    private static class PollSet {  
3.        protected IPollEvents handler;   //事件的回调  
4.        protected SelectionKey key;   //注册以后的key  
5.        protected int ops;    //注册的事件  
6.        protected boolean cancelled;   //是否已经取消  
7.          
8.        protected PollSet(IPollEvents handler) {  
9.            this.handler = handler;  
10.            key = null;  
11.            cancelled = false;  
12.            ops = 0;  
13.        }  
14.    }  
15.    final private Map<SelectableChannel, PollSet> fd_table;   //记录全部的注册，key是channel  
16.   
17.    //  If true, there's at least one retired event source.  
18.    private boolean retired;    //当前注册的对象是否有更新，若是有更新的话，在执行select以前须要先更新注册  
19.   
20.    //  If true, thread is in the process of shutting down.  
21.    volatile private boolean stopping;    //若是是true的话，那么执行线程将会中止  
22.    volatile private boolean stopped;   //是否已经中止  
23.      
24.    private Thread worker;   //worker线程  
25.    private Selector selector;   //selector  
26.    final private String name;   //名字  

这里显示定义了一个嵌套类，全部须要注册到selector上的channel都会先构建这个对象，将其当作附件注册到selector上。。。。其中handler是事件回调，key是selector注册后取得的key，ops是注册的事件类型

而后是fd_table，这个应该知道是干吗用的吧，用于关联注册的channel对象与其的PollSet对象。。。

这里的retired用于标识当前的注册的channel什么的是否有更新。。。接下来的重要属性还有thread，这个是干吗应该很清楚吧，还有一个selector就很少说了。。。

接下来来看看如何在poller对象上面注册channel吧，有几个比较重要的方法：

[java]  view plain copy

1. //用于在当前的集合里面添加须要注册的channel，第一个参数是channel，第二个参数是事件回调  
2. public final void add_fd (SelectableChannel fd_, IPollEvents events_) {  
3.     fd_table.put(fd_, new PollSet(events_));  //直接把放到map里面就行了  
4.     adjust_load (1);  //增长load值，这里所谓的负载其实就是在当前poller里面注册的channel的数量  
5. }  
6. //在key上面注册事件，若是negate为true的话，那么表示是取消事件  
7. private final void register (SelectableChannel handle_, int ops, boolean negate) {  
8.     PollSet pollset = fd_table.get(handle_);  //获取pollset对象  
9.       
10.     if (negate)  {  
11.         pollset.ops = pollset.ops &~ ops;  //取反，至关于取消事件  
12.     } else {  
13.         pollset.ops = pollset.ops | ops;  //注册事件  
14.     }  
15.       
16.     if (pollset.key != null) {  //若是有key了，那么表示已经注册到selector上面了，那么只须要更新key就行了  
17.         pollset.key.interestOps(pollset.ops);    
18.     } else {  
19.         retired = true;  
20.   
21.     }  
22. }  

这里首先须要调用add_fd方法，channel加入进去，而后再调用register方法注册相应的事件，不知道为啥要这么弄。。直接一个方法实现不就行了么。。可能有一些细节的东西我还不太清楚吧，很少说这个了。。

好了，接下来来看看它的run方法吧：

[java]  view plain copy

1. //poller的执行流程  
2. public void run () {  
3.     int returnsImmediately = 0;  
4.   
5.     while (!stopping) {  
6.         long timeout = execute_timers ();  //执行全部的超时，而且获取下一个超时的时间  
7.           
8.         if (retired) {  //这里表示注册的东西有更新  
9.               
10.             Iterator <Map.Entry <SelectableChannel,PollSet>> it = fd_table.entrySet ().iterator ();  
11.             while (it.hasNext ()) {  //遍历全部须要注册的  
12.                 Map.Entry <SelectableChannel,PollSet> entry = it.next ();  
13.                 SelectableChannel ch = entry.getKey ();  //获取channel  
14.                 PollSet pollset = entry.getValue ();   //获取pollset  
15.                 if (pollset.key == null) {  //这里没有key的话，表示当前channel并无注册到selector上面去  
16.                     try {  
17.                         pollset.key = ch.register(selector, pollset.ops, pollset.handler);   //注册，这里注册的附件竟然是事件的回调函数  
18.                     } catch (ClosedChannelException e) {  
19.                     }  
20.                 }   
21.                   
22.                   
23.                 if (pollset.cancelled || !ch.isOpen()) {  //若是是取消注册，那么直接取消掉就能够了  
24.                     if(pollset.key != null) {  
25.                         pollset.key.cancel();  
26.                     }  
27.                     it.remove ();  
28.                 }  
29.             }  
30.             retired = false;  
31.               
32.         }  
33.   
34.         //  Wait for events.  
35.         int rc;  
36.         long start = System.currentTimeMillis ();  //select以前的时间  
37.         try {  
38.             rc = selector.select (timeout);  
39.         } catch (IOException e) {  
40.             throw new ZError.IOException (e);  
41.         }  
42.           
43.         if (rc == 0) {   //出错啦，好像  
44.             //  Guess JDK epoll bug  
45.             if (timeout == 0 ||  
46.                     System.currentTimeMillis () - start < timeout / 2)  
47.                 returnsImmediately ++;  
48.             else  
49.                 returnsImmediately = 0;  
50.   
51.             if (returnsImmediately > 10) {  
52.                 rebuildSelector ();   //重建selector  
53.                 returnsImmediately = 0;  
54.             }  
55.             continue;  
56.         }  
57.   
58.   
59.         Iterator<SelectionKey> it = selector.selectedKeys().iterator();  //全部select出来的key  
60.         while (it.hasNext()) {  //遍历  
61.             SelectionKey key = it.next();  
62.             IPollEvents evt = (IPollEvents) key.attachment();  
63.             it.remove();  
64.   
65.             try {  //接下来就是判断事件的类型执行相应的方法就行了  
66.                 if (key.isReadable() ) {  //有数据能够读取了   
67.                     evt.in_event();  
68.                 } else if (key.isAcceptable()) {  //有新的链接进来了  
69.                     evt.accept_event();  
70.                 } else if (key.isConnectable()) {  //链接创建  
71.                     evt.connect_event();  
72.                 }   
73.                 if (key.isWritable()) {  //可写  
74.                     evt.out_event();  
75.                 }   
76.             } catch (CancelledKeyException e) {  
77.                 // channel might have been closed  
78.             }  
79.               
80.         }  
81.   
82.     }  
83.       
84.     stopped = true;  
85.       
86. }  

这个应该很容易看懂吧，首先执行了全部超时的事件，而后若是有注册的channel更新的话，须要从新更新这些注册，而后就能够执行select方法了，接着遍历出全部select的key，而后判断事件的类型，执行相应的回调方法就行了。。。

最后来看看它的start方法：

[java]  view plain copy

1. //启动，这里主要是建立一个线程，而后开始运行  
2. public void start() {  
3.     worker = new Thread(this, name);  //建立thread，  
4.     worker.start();  //启动这个执行线程  
5. }  

好吧，简单吧，建立一个线程，而后启动就行了，这里执行的就是run方法。。。。

好了，到这里整个poller的实现和其运行基本上就算是搞清楚了。。。并且能够知道poller对象才是真的I/O线程的持有者。。。。

接下来来介绍另一个类型：Mailbox，每个I/O线程都会有本身的mailbox，并且链接也会有本身的mailbox，能够向mailbox里面发送命令，而后让其执行。。。这里能够理解为mailbox是命令的接收器，ZeroMQ就是用这个来实现组件之间的通讯的。。。。

先来看看他的一些重要的属性定义吧：

[java]  view plain copy

1. private final YPipe<Command> cpipe;   //这名字太唬人了，其实就是一个保存command的队列而已  
2.   
3. //其实能够将其理解为一个socketpair，若是有命令写入了队列，那么经过在这里写入一个数据，能够用于提醒有命令发送到了mialbox  
4. private final Signaler signaler;   //用于通讯的signal，使用pipe实现的。。。，其实这里只不过是一个噱头，这里写入数据是为了提醒执行线程command队列里面有命令写入了  
5.   
6. private final Lock sync;  //只有一个线程从mailbox里面收命令，可是会有不少线程向mialbox里面发送命令，用这个锁来保护  
7.   
8. private boolean active;   //用于判断底层的pipe是否仍是活跃的，若是是true的话，表示底层的pipe活跃，能够读取命令  
9.   
10. // mailbox name, for better debugging  
11. private final String name;   //当前mailbox的名字  

这里cpipe这个名字比较唬人，其实能够就将其理解为一个command的队列，全部的命令都会放到这个里面去，而后是signaler，这个是底层通讯的实现，它里面建立了pipe，相似于socketpair，经过在在这个里面写数据，用于提醒cpipe里面有命令写进去了。。须要处理。。。

来看看几个比较重要的方法吧：

[java]  view plain copy

1. public SelectableChannel get_fd () {  
2.     return signaler.get_fd ();   //这里其实获取的是signal用到的pipe的读channel  
3. }  
4.   
5. //向当前的mailbox发送命令，其实就是写到command队列里面去而已  
6. public void send (final Command cmd_) {     
7.     boolean ok = false;  
8.     sync.lock ();  
9.     try {  
10.         cpipe.write (cmd_, false);  
11.         ok = cpipe.flush ();  //pipeflush，这里将会被selector感应到，从而能够执行相应的处理，在执行线程里面执行命令  
12.     } finally {  
13.         sync.unlock ();  
14.     }  
15.       
16.     if (!ok) {  
17.         signaler.send (); //经过写端写数据，这样子的话会被读端收到  
18.     }  
19. }  
20.   
21. //收取命令，若是这里没法马上获取命令的话，还能够有一个超时时间  
22. public Command recv (long timeout_)  {  
23.     Command cmd_ = null;  
24.     //  Try to get the command straight away.  
25.     if (active) {  
26.         cmd_ = cpipe.read ();  //从队列里面获取命令  
27.         if (cmd_ != null) {  
28.               
29.             return cmd_;  
30.         }  
31.         //  If there are no more commands available, switch into passive state.  
32.         active = false;  
33.         signaler.recv ();  //这里会从读端不断的读数据  
34.     }  
35.   
36.   
37.     //  Wait for signal from the command sender.  
38.     boolean rc = signaler.wait_event (timeout_);  
39.     if (!rc)  
40.         return null;  
41.   
42.     //  We've got the signal. Now we can switch into active state.  
43.     active = true;  
44.   
45.     //  Get a command.  
46.     cmd_ = cpipe.read ();  
47.     assert (cmd_ != null);  
48.       
49.     return cmd_;  
50. }  

这里获取底层的fd，其实就是获取用于通讯的signal的读端的channel，而后向这个mailbox发送命令其实就是直接向command的队列里面放入命令就行了，而且这里须要经过signaler来提醒一下。。。。

而后recv方法，用于获取命令，其实最终仍是在命令队列里去拿。。。。

好了，到这里mailbox差很少了，一些细节并无贴出来，由于其实这东西若是没有搞懂具体是怎么用的话也不可能搞得明白。。。。

好了，在最后开始IOThread这个类型以前先来介绍另外两个东西吧：

（1）IPollEvents，这个是一个接口，也就是事件的回调。。来看看它的定义就知道了。。。

[java]  view plain copy

1. public interface IPollEvents {  
2.     void in_event () ;  //当有数据能够读取的时候须要执行的方法  
3.     void out_event () ;  //当能够写的时候应该执行的方法  
4.     void connect_event () ;  //当已经创建了链接以后，应该执行的  
5.     void accept_event();  //当有accept的时候，应该执行这个  
6.     void timer_event (int id_) ;  //当超时的时候应该执行的  
7. }  

里面定义了5个方法，具体这5个方法分别处理什么事件应该看名字就可以很容易知道吧。。就不细说了。。

（2）ZObject，这个类型是干吗的呢，在前面已已经说过了，mailbox用于存取别的地方发送过来的命令，而ZObject就是用于执行命令的，若是须要组件能够进行命令的交互，那么就须要类型实现继承ZObject，具体的类容就不说了，有兴趣的本身看吧，很简单的，，，，

好啦，终于到了最激动人心的时候了，来看看IOThread类型，看这个名字就知道它是干吗的吧，先来看看它的类型定义图吧：

其实看到这里也可以猜出来IOThread类型自己并无太多的内容，更多的时候都是有mailbox，poller来作了。。。

来看看它的一些重要属性和构造函数吧：

[java]  view plain copy

1. final private Mailbox mailbox;   //I/O线程将会从这个mailbox里面获取命令  
2.   
3. final private SelectableChannel mailbox_handle;    //mailbox会用到的chanel，其实也就是底层pipe的读端  
4.   
5. final private Poller poller;  //poller对象  
6.   
7. final String name;  //这个IO线程的名字  
8.   
9. public IOThread(Ctx ctx_, int tid_) {  //所属的ctx，以及这个是第几个IO线程，也能够把它理解为ID吧  
10.     super(ctx_, tid_);  
11.     name = "iothread-" + tid_;  
12.     poller = new Poller(name);  //建立poller  
13.   
14.     mailbox = new Mailbox(name);  //建立mailbox  
15.     mailbox_handle = mailbox.get_fd();  //mailbox会用到的channel，pipe的读端  
16.     poller.add_fd (mailbox_handle, this);   //在poller里面注册，其实这里只是将其放到fd列表里面，这里的事件回调就是当前对象  
17.     poller.set_pollin (mailbox_handle);  //这里注册读取事件   
18. }  

这里mailbox和poller是干吗用的就很少说了，另外这个mailbox_handle实际上是mailbox的signaler的读端，并且能够在构造函数中能够看到将这个channel注册到了poller上面去。。这样若是有数据读，那么会被响应，也就意味着有命令发送到mailbox须要执行了。。。

咱们来看看这个回调函：

[java]  view plain copy

1. //当mailbox能够读取的时候，将会执行这个方法，这里其实也就是收到了命令  
2. public void in_event() {  
3.     //  TODO: Do we want to limit number of commands I/O thread can  
4.     //  process in a single go?  
5.   
6.     while (true) {  
7.   
8.         //  Get the next command. If there is none, exit.  
9.         //获取须要执行的命令  
10.         Command cmd = mailbox.recv (0);  
11.         if (cmd == null)  
12.             break;  
13.   
14.         //  Process the command.  
15.         //执行命令  
16.         cmd.destination().process_command (cmd);  //其实对于IO线程对象，也就只有stop命令能够执行  
17.     }  
18.   
19. }  

简单吧，从mailbox里面获取command，而后直接执行就行了。。。。这里IOThread自己就继承了ZOjbect，因此这里说白了就是本身须要执行命令，而在IOThread中，只有stop命令须要执行：

[java]  view plain copy

1. //中止poller  
2. protected void process_stop ()  
3. {  
4.     poller.rm_fd (mailbox_handle);  
5.       
6.     poller.stop ();  
7.   
8. }  

好啦，到这里ZeroMQ中IO线程的实现应该就算是比较的清楚了。。并且如何实现组件间的通讯也算是比较的了解了。。。