算是开始读ZeroMQ(java)的代码实现了吧,现在有了一个大体的了解,看起来实现是比较的干净的,抽象什么的不算复杂。。。
这里先来看看它的I/O线程的实现吧,顺带看看是如何实现组件的通信的。。。。
首先要搞清楚I/O线程的实现,就先要弄懂一个类型,Poller(zmq.Poller.java),可以将其看成是对selector的一个封装,同时它还要管理定时事件,看了这么多代码,发现基本上都是在实现I/Oselect的地方完成了定时的实现。。。。
好了,不说太多闲话了,来看看它的继承体系吧:
这里还将依赖关系也标出来了,首先继承自PollerBase抽象类,然后实现了Runnable接口,自己还会创建一个Thread对象。。。看了这个图,基本上就已经能够知道Poller的运行原理了吧。。。。
这里先来看看PollerBase的实现吧,它其实主要是用来管理定时的,那么先来看看他的一些重要的属性和定义:
[java] view plain copy private final AtomicInteger load; //这个load其实就是当前poller里面注册的channel的数量 //这里是要注册的超时是事件 private final class TimerInfo { IPollEvents sink; //事件回调 int id; public TimerInfo(IPollEvents sink_, int id_) { sink = sink_; id = id_; } } private final Map<Long, TimerInfo> timers; //这里记录所有的超时对象,key是时间 private final Map<Long, TimerInfo> addingTimers; //等待加入的超时事件前面的一个原子Integer是用于记录负载的,用于记录当前poller里面一共注册了多少I/O对象。。。然后是超时事件的定义,sink是超时的事件回调函数,里面有相应的方法,timer就记录了所有的超时事件,addingTimers是需要加入的超时事件。。这里的key都是超时的时间,value就是超时对象了。。。
这里就来看两个主要的方法就好了吧,先来看看如何加入超时事件:
[java] view plain copy //添加一个超时事件 public void add_timer (long timeout_, IPollEvents sink_, int id_) { long expiration = Clock.now_ms () + timeout_; //计算超时的时间 TimerInfo info = new TimerInfo(sink_, id_); //创建超时对象 addingTimers.put(expiration, info); //将其添加到adding里面去 }代码应该很简单能够看明白吧,第一个参数是超时时间,第二个参数是回调方法,第三个参数是ID,首先加上当前的时间就算出了超时的时间,然后创建超时对象,这里先是将其放入了addingTimers里面,而不是直接放到了timer里面,。。。
那么接下来来看看如何执行所有的超时的方法吧:
[java] view plain copy //执行所有的超时事件,返回下一个超时还剩下的时间 protected long execute_timers() { if (!addingTimers.isEmpty()) { //如果当前还有需要添的超时时间,那么需要将其添加进去 timers.putAll(addingTimers); addingTimers.clear(); } //没有超时事件 if (timers.isEmpty()) return 0L; //获取当前的时间 long current = Clock.now_ms (); //遍历所有的超时时间,这里是从最小的开始的 Iterator<Entry <Long, TimerInfo>> it = timers.entrySet().iterator(); while (it.hasNext()) { Entry <Long, TimerInfo> o = it.next(); // If we have to wait to execute the item, same will be true about // all the following items (multimap is sorted). Thus we can stop // checking the subsequent timers and return the time to wait for // the next timer (at least 1ms). //如果超时的时间大于当前的时间,那么表示还没有超时, if (o.getKey() > current) { return o.getKey() - current; //返回下一个超时还剩下的时间 } // Trigger the timer. //执行超时方法 o.getValue().sink.timer_event (o.getValue().id); // Remove it from the list of active timers. it.remove(); } if (!addingTimers.isEmpty()) return execute_timers(); // There are no more timers. return 0L; //如果是0 的话,表示没有timer执行了 } }应该代码也还算比较好理解吧,这里可以看到将addingTimers里面的都放到了timers里面。。。然后遍历所有的超时对象,并执行他们的超时回调,知道一个超时时间还没有到,最后返回的是下一个超时事件还剩下多长的时间。。。
好了,那么接下来来看看Poller类型的实现吧,先来看看它的重要定义:
[java] view plain copy //在当前poller里面注册的封装。。。 private static class PollSet { protected IPollEvents handler; //事件的回调 protected SelectionKey key; //注册之后的key protected int ops; //注册的事件 protected boolean cancelled; //是否已经取消 protected PollSet(IPollEvents handler) { this.handler = handler; key = null; cancelled = false; ops = 0; } } final private Map<SelectableChannel, PollSet> fd_table; //记录所有的注册,key是channel // If true, there's at least one retired event source. private boolean retired; //当前注册的对象是否有更新,如果有更新的话,在执行select之前需要先更新注册 // If true, thread is in the process of shutting down. volatile private boolean stopping; //如果是true的话,那么执行线程将会停止 volatile private boolean stopped; //是否已经停止 private Thread worker; //worker线程 private Selector selector; //selector 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 //用于在当前的集合里面添加需要注册的channel,第一个参数是channel,第二个参数是事件回调 public final void add_fd (SelectableChannel fd_, IPollEvents events_) { fd_table.put(fd_, new PollSet(events_)); //直接把放到map里面就好了 adjust_load (1); //增加load值,这里所谓的负载其实就是在当前poller里面注册的channel的数量 } //在key上面注册事件,如果negate为true的话,那么表示是取消事件 private final void register (SelectableChannel handle_, int ops, boolean negate) { PollSet pollset = fd_table.get(handle_); //获取pollset对象 if (negate) { pollset.ops = pollset.ops &~ ops; //取反,相当于取消事件 } else { pollset.ops = pollset.ops | ops; //注册事件 } if (pollset.key != null) { //如果有key了,那么表示已经注册到selector上面了,那么只需要更新key就好了 pollset.key.interestOps(pollset.ops); } else { retired = true; } }这里首先需要调用add_fd方法,channel加入进去,然后再调用register方法注册相应的事件,不知道为啥要这么弄。。直接一个方法实现不就好了么。。可能有一些细节的东西我还不太清楚吧,不多说这个了。。
好了,接下来来看看它的run方法吧:
[java] view plain copy //poller的执行流程 public void run () { int returnsImmediately = 0; while (!stopping) { long timeout = execute_timers (); //执行所有的超时,并且获取下一个超时的时间 if (retired) { //这里表示注册的东西有更新 Iterator <Map.Entry <SelectableChannel,PollSet>> it = fd_table.entrySet ().iterator (); while (it.hasNext ()) { //遍历所有需要注册的 Map.Entry <SelectableChannel,PollSet> entry = it.next (); SelectableChannel ch = entry.getKey (); //获取channel PollSet pollset = entry.getValue (); //获取pollset if (pollset.key == null) { //这里没有key的话,表示当前channel并没有注册到selector上面去 try { pollset.key = ch.register(selector, pollset.ops, pollset.handler); //注册,这里注册的附件居然是事件的回调函数 } catch (ClosedChannelException e) { } } if (pollset.cancelled || !ch.isOpen()) { //如果是取消注册,那么直接取消掉就可以了 if(pollset.key != null) { pollset.key.cancel(); } it.remove (); } } retired = false; } // Wait for events. int rc; long start = System.currentTimeMillis (); //select之前的时间 try { rc = selector.select (timeout); } catch (IOException e) { throw new ZError.IOException (e); } if (rc == 0) { //出错啦,好像 // Guess JDK epoll bug if (timeout == 0 || System.currentTimeMillis () - start < timeout / 2) returnsImmediately ++; else returnsImmediately = 0; if (returnsImmediately > 10) { rebuildSelector (); //重建selector returnsImmediately = 0; } continue; } Iterator<SelectionKey> it = selector.selectedKeys().iterator(); //所有select出来的key while (it.hasNext()) { //遍历 SelectionKey key = it.next(); IPollEvents evt = (IPollEvents) key.attachment(); it.remove(); try { //接下来就是判断事件的类型执行相应的方法就好了 if (key.isReadable() ) { //有数据可以读取了 evt.in_event(); } else if (key.isAcceptable()) { //有新的连接进来了 evt.accept_event(); } else if (key.isConnectable()) { //连接建立 evt.connect_event(); } if (key.isWritable()) { //可写 evt.out_event(); } } catch (CancelledKeyException e) { // channel might have been closed } } } stopped = true; }这个应该很容易看懂吧,首先执行了所有超时的事件,然后如果有注册的channel更新的话,需要重新更新这些注册,然后就可以执行select方法了,接着遍历出所有select的key,然后判断事件的类型,执行相应的回调方法就好了。。。
最后来看看它的start方法:
[java] view plain copy //启动,这里主要是创建一个线程,然后开始运行 public void start() { worker = new Thread(this, name); //创建thread, worker.start(); //启动这个执行线程 }好吧,简单吧,创建一个线程,然后启动就好了,这里执行的就是run方法。。。。
好了,到这里整个poller的实现和其运行基本上就算是搞清楚了。。。而且可以知道poller对象才是真的I/O线程的持有者。。。。
接下来来介绍另外一个类型:Mailbox,每一个I/O线程都会有自己的mailbox,而且连接也会有自己的mailbox,可以向mailbox里面发送命令,然后让其执行。。。这里可以理解为mailbox是命令的接收器,ZeroMQ就是用这个来实现组件之间的通信的。。。。
先来看看他的一些重要的属性定义吧:
[java] view plain copy private final YPipe<Command> cpipe; //这名字太唬人了,其实就是一个保存command的队列而已 //其实可以将其理解为一个socketpair,如果有命令写入了队列,那么通过在这里写入一个数据,可以用于提醒有命令发送到了mialbox private final Signaler signaler; //用于通信的signal,使用pipe实现的。。。,其实这里只不过是一个噱头,这里写入数据是为了提醒执行线程command队列里面有命令写入了 private final Lock sync; //只有一个线程从mailbox里面收命令,但是会有很多线程向mialbox里面发送命令,用这个锁来保护 private boolean active; //用于判断底层的pipe是否还是活跃的,如果是true的话,表示底层的pipe活跃,可以读取命令 // mailbox name, for better debugging private final String name; //当前mailbox的名字这里cpipe这个名字比较唬人,其实可以就将其理解为一个command的队列,所有的命令都会放到这个里面去,然后是signaler,这个是底层通信的实现,它里面创建了pipe,类似于socketpair,通过在在这个里面写数据,用于提醒cpipe里面有命令写进去了。。需要处理。。。
来看看几个比较重要的方法吧:
[java] view plain copy public SelectableChannel get_fd () { return signaler.get_fd (); //这里其实获取的是signal用到的pipe的读channel } //向当前的mailbox发送命令,其实就是写到command队列里面去而已 public void send (final Command cmd_) { boolean ok = false; sync.lock (); try { cpipe.write (cmd_, false); ok = cpipe.flush (); //pipeflush,这里将会被selector感应到,从而可以执行相应的处理,在执行线程里面执行命令 } finally { sync.unlock (); } if (!ok) { signaler.send (); //通过写端写数据,这样子的话会被读端收到 } } //收取命令,如果这里无法立刻获取命令的话,还可以有一个超时时间 public Command recv (long timeout_) { Command cmd_ = null; // Try to get the command straight away. if (active) { cmd_ = cpipe.read (); //从队列里面获取命令 if (cmd_ != null) { return cmd_; } // If there are no more commands available, switch into passive state. active = false; signaler.recv (); //这里会从读端不断的读数据 } // Wait for signal from the command sender. boolean rc = signaler.wait_event (timeout_); if (!rc) return null; // We've got the signal. Now we can switch into active state. active = true; // Get a command. cmd_ = cpipe.read (); assert (cmd_ != null); return cmd_; }这里获取底层的fd,其实就是获取用于通信的signal的读端的channel,然后向这个mailbox发送命令其实就是直接向command的队列里面放入命令就好了,并且这里需要通过signaler来提醒一下。。。。
然后recv方法,用于获取命令,其实最终还是在命令队列里去拿。。。。
好了,到这里mailbox差不多了,一些细节并没有贴出来,因为其实这东西如果没有搞懂具体是怎么用的话也不可能搞得明白。。。。
好了,在最后开始IOThread这个类型之前先来介绍另外两个东西吧:
(1)IPollEvents,这个是一个接口,也就是事件的回调。。来看看它的定义就知道了。。。
[java] view plain copy public interface IPollEvents { void in_event () ; //当有数据可以读取的时候需要执行的方法 void out_event () ; //当可以写的时候应该执行的方法 void connect_event () ; //当已经建立了连接之后,应该执行的 void accept_event(); //当有accept的时候,应该执行这个 void timer_event (int id_) ; //当超时的时候应该执行的 }里面定义了5个方法,具体这5个方法分别处理什么事件应该看名字就能够很容易知道吧。。就不细说了。。
(2)ZObject,这个类型是干嘛的呢,在前面已已经说过了,mailbox用于存取别的地方发送过来的命令,而ZObject就是用于执行命令的,如果需要组件可以进行命令的交互,那么就需要类型实现继承ZObject,具体的类容就不说了,有兴趣的自己看吧,很简单的,,,,
好啦,终于到了最激动人心的时候了,来看看IOThread类型,看这个名字就知道它是干嘛的吧,先来看看它的类型定义图吧:
其实看到这里也能够猜出来IOThread类型本身并没有太多的内容,更多的时候都是有mailbox,poller来做了。。。
来看看它的一些重要属性和构造函数吧:
[java] view plain copy final private Mailbox mailbox; //I/O线程将会从这个mailbox里面获取命令 final private SelectableChannel mailbox_handle; //mailbox会用到的chanel,其实也就是底层pipe的读端 final private Poller poller; //poller对象 final String name; //这个IO线程的名字 public IOThread(Ctx ctx_, int tid_) { //所属的ctx,以及这个是第几个IO线程,也可以把它理解为ID吧 super(ctx_, tid_); name = "iothread-" + tid_; poller = new Poller(name); //创建poller mailbox = new Mailbox(name); //创建mailbox mailbox_handle = mailbox.get_fd(); //mailbox会用到的channel,pipe的读端 poller.add_fd (mailbox_handle, this); //在poller里面注册,其实这里只是将其放到fd列表里面,这里的事件回调就是当前对象 poller.set_pollin (mailbox_handle); //这里注册读取事件 }这里mailbox和poller是干嘛用的就不多说了,另外这个mailbox_handle其实是mailbox的signaler的读端,而且可以在构造函数中可以看到将这个channel注册到了poller上面去。。这样如果有数据读,那么会被响应,也就意味着有命令发送到mailbox需要执行了。。。
我们来看看这个回调函:
[java] view plain copy //当mailbox可以读取的时候,将会执行这个方法,这里其实也就是收到了命令 public void in_event() { // TODO: Do we want to limit number of commands I/O thread can // process in a single go? while (true) { // Get the next command. If there is none, exit. //获取需要执行的命令 Command cmd = mailbox.recv (0); if (cmd == null) break; // Process the command. //执行命令 cmd.destination().process_command (cmd); //其实对于IO线程对象,也就只有stop命令可以执行 } }简单吧,从mailbox里面获取command,然后直接执行就好了。。。。这里IOThread本身就继承了ZOjbect,所以这里说白了就是自己需要执行命令,而在IOThread中,只有stop命令需要执行:
[java] view plain copy //停止poller protected void process_stop () { poller.rm_fd (mailbox_handle); poller.stop (); }好啦,到这里ZeroMQ中IO线程的实现应该就算是比较的清楚了。。而且如何实现组件间的通信也算是比较的了解了。。。
转载于:https://www.cnblogs.com/jym-sunshine/p/5441774.html
相关资源:数据结构—成绩单生成器