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

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

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

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

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

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

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

[java] view plaincopy

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

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

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

[java] view plaincopy

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. }

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

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

[java] view plaincopy

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. long current = Clock.now_ms ();
12. //遍历所有的超时时间，这里是从最小的开始的
13. Iterator<Entry <Long, TimerInfo>> it = timers.entrySet().iterator();
14. while (it.hasNext()) {
15. Entry <Long, TimerInfo> o = it.next();
16. //  If we have to wait to execute the item, same will be true about
17. //  all the following items (multimap is sorted). Thus we can stop
18. //  checking the subsequent timers and return the time to wait for
19. //  the next timer (at least 1ms).
20. //如果超时的时间大于当前的时间，那么表示还没有超时，
21. if (o.getKey() > current) {
22. return o.getKey() - current;  //返回下一个超时还剩下的时间
23. }
24. //  Trigger the timer.
25. //执行超时方法
26. o.getValue().sink.timer_event (o.getValue().id);
27. //  Remove it from the list of active timers.
28. it.remove();
29. }
30. if (!addingTimers.isEmpty())
31. return execute_timers();
32. //  There are no more timers.
33. return 0L;  //如果是0 的话，表示没有timer执行了
34. }
35. }

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

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

[java] view plaincopy

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. protected PollSet(IPollEvents handler) {
8. this.handler = handler;
9. key = null;
10. cancelled = false;
11. ops = ;
12. }
13. }
14. final private Map<SelectableChannel, PollSet> fd_table;   //记录所有的注册，key是channel
15. //  If true, there's at least one retired event source.
16. private boolean retired;    //当前注册的对象是否有更新，如果有更新的话，在执行select之前需要先更新注册
17. //  If true, thread is in the process of shutting down.
18. volatile private boolean stopping;    //如果是true的话，那么执行线程将会停止
19. volatile private boolean stopped;   //是否已经停止
20. private Thread worker;   //worker线程
21. private Selector selector;   //selector
22. final private String name;   //名字

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

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

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

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

[java] view plaincopy

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 ();  //增加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. if (negate)  {
10. pollset.ops = pollset.ops &~ ops;  //取反，相当于取消事件
11. } else {
12. pollset.ops = pollset.ops | ops;  //注册事件
13. }
14. if (pollset.key != null) {  //如果有key了，那么表示已经注册到selector上面了，那么只需要更新key就好了
15. pollset.key.interestOps(pollset.ops);
16. } else {
17. retired = true;
18. }
19. }

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

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

[java] view plaincopy

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

这个应该很容易看懂吧，首先执行了所有超时的事件，然后如果有注册的channel更新的话，需要重新更新这些注册，然后就可以执行select方法了，接着遍历出所有select的key，然后判断事件的类型，执行相应的回调方法就好了。。。

最后来看看它的start方法：

[java] view plaincopy

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 plaincopy

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

这里cpipe这个名字比较唬人，其实可以就将其理解为一个command的队列，所有的命令都会放到这个里面去，然后是signaler，这个是底层通信的实现，它里面创建了pipe，类似于socketpair，通过在在这个里面写数据，用于提醒cpipe里面有命令写进去了。。需要处理。。。

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

[java] view plaincopy

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

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

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

好了，到这里mailbox差不多了，一些细节并没有贴出来，因为其实这东西如果没有搞懂具体是怎么用的话也不可能搞得明白。。。。

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

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

[java] view plaincopy

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 plaincopy

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

这里mailbox和poller是干嘛用的就不多说了，另外这个mailbox_handle其实是mailbox的signaler的读端，而且可以在构造函数中可以看到将这个channel注册到了poller上面去。。这样如果有数据读，那么会被响应，也就意味着有命令发送到mailbox需要执行了。。。

我们来看看这个回调函：

[java] view plaincopy

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. while (true) {
6. //  Get the next command. If there is none, exit.
7. //获取需要执行的命令
8. Command cmd = mailbox.recv ();
9. if (cmd == null)
10. break;
11. //  Process the command.
12. //执行命令
13. cmd.destination().process_command (cmd);  //其实对于IO线程对象，也就只有stop命令可以执行
14. }
15. }

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

[java] view plaincopy

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

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