这里cpipe这个名字比较唬人，其实可以就将其理解为一个command的队列，所有的命令都会放到这个里面去，然后是signaler，这个是底层通信的实现，它里面创建了pipe，类似于socketpair，通过在在这个里面写数据，用于提醒cpipe里面有命令写进去了。。需要处理。。。
来看看几个比较重要的方法吧：
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，这个是一个接口，也就是事件的回调。。来看看它的定义就知道了。。。
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来做了。。。< http://www.2cto.com/kf/ware/vc/" target="_blank" class="keylink">vcD4KPHA+wLS/tL+0y/y1xNK70KnW2NKqyvTQ1LrNubnU7Lqvyv2wyaO6PHByZSBjbGFzcz0="brush:java;"> 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需要执行了。。。
我们来看看这个回调函：
//当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命令需要执行：
//停止poller
protected void process_stop ()
{
poller.rm_fd (mailbox_handle);
poller.stop ();
}
好啦，到这里ZeroMQ中IO线程的实现应该就算是比较的清楚了。。而且如何实现组件间的通信也算是比较的了解了。。。