标签：
在zeromq源码分析笔记之架构说到了zmq的整体架构，可以看到线程间通信包括两类，一类是用于收发命令，告知对象该调用什么方法去做什么事情，命令的结构由command_t结构体确定；另一类是socket_base_t实例与session的消息通信，消息的结构由msg_t确定。命令的发送与存储是通过mailbox_t实现的，消息的发送和存储是通过pipe_t实现的，这两个结构都会详细说到，今天先说一下线程间的收发命令。
zeromq的线程可分为两类，一类是io线程，像reaper_t、io_thread_t都属于这一类，这类线程的特点就是内含一个轮询器poller及mailbox_t，通过poller可以监听mailbox_t的信件到来 ；另一类是zmq的socket,所有socket_base_t实例化的对象都可以看做一个单独的线程，这类线程不含poller，但同样含有一个mailbox_t，可以用于收发命令，由于不含poller，只能在每次使用socket_base_t实例的时候先处理一下mailbox_t，看是否有命令需要处理，代码上来看就是每次先调用下面这个函数接收并处理一下命令：
int zmq::socket_base_t::process_commands (int timeout_, bool throttle_)
另外，两类线程发送命令的方式是一致的。下面，就详细的说一下命令结构、如何发送命令、两类线程如何接收命令
先看一下命令结构（详细的结构参见源码Command.hpp）：
// This structure defines the commands that can be sent between threads.
struct command_t
{
// Object to process the command.
zmq::object_t *destination;
enum type_t
{
...
} type;
union {
...
} args;
};
可以看到，命令由三部分构成，分别是发往的目的地destination，命令的类型type，命令的参数args。所谓的命令就是一个对象交代另一个对象去做某件事情，说白了就是告诉令一个对象应该调用哪个方法，命令的发出者是一个对象，而接收者是一个线程，线程接收到命令后，根据目的地派发给相应的对象做处理。可以看到命令的destination属性是object_t类型的，在上节介绍类的层次结构图时，说到object_t及其子类都具有发送和处理命令的功能（没有收命令的功能），所以有必要弄清楚一件事，对象、object_t、poller、线程、mailbox_t、命令是什么关系？
简单来说就是，object_t发命令，poller监听命令到来告知线程收命令，交给object_t处理，无论是object_t还是poller其实都操作mailbox_t，而这三者都在绑定在同一个线程上。下面就来看看具体的如何发送命令
一个对象想使用线程的发命令功能，其类就得继承自object_t（源码在Object.hpp/.cpp）：
class object_t { public: object_t (zmq::ctx_t *ctx_, uint32_t tid_); void process_command (zmq::command_t &cmd_); ... protected: ... private: zmq::ctx_t *ctx;// Context provides access to the global state. uint32_t tid;// Thread ID of the thread the object belongs to. void send_command (command_t &cmd_); }
可以看到，object_t内含一个tid，含义就是，该object_t对象要使用哪个线程的mailbox_t。关于zmq::ctx_t，在zmq中被称为上下文语境，上下文语境简单来说就是zmq的存活环境，里面存储是一些全局对象，zmq中所有的线程都可以使用这些对象。zmq线程中的mailbox_t对象会被zmq存储在ctx_t对象中。zmq的做法就是，在上下文语境中使用一个容器slots装载线程的mailbox，在新建线程的时候，给线程分配一个线程标志tid和mailbox，把mailbox放入容器的tid那个位置，代码来说就是slots[tid]=mailbox。有了这个基础，线程A给线程B发命令就只要往slots[B.tid]写入命令就可以了：
void zmq::object_t::send_command (command_t &cmd_) { ctx->send_command (cmd_.destination->get_tid (), cmd_); } void zmq::ctx_t::send_command (uint32_t tid_, const command_t &command_) { slots [tid_]->send (command_); } void zmq::mailbox_t::send (const command_t &cmd_) { sync.lock(); cpipe.write (cmd_, false); bool ok = cpipe.flush (); sync.unlock (); if (!ok) signaler.send (); }
前面说过，每个io线程都含有一个poller，io线程的结构如下（源码在Io_thread_t.hpp/.cpp）：
class io_thread_t : public object_t, public i_poll_events { public: io_thread_t (zmq::ctx_t *ctx_, uint32_t tid_); ~io_thread_t (); void start (); // Launch the physical thread. void stop ();// Ask underlying thread to stop. ... private: mailbox_t mailbox;// I/O thread accesses incoming commands via this mailbox. poller_t::handle_t mailbox_handle;// Handle associated with mailbox‘ file descriptor. poller_t *poller;// I/O multiplexing is performed using a poller object. } zmq::io_thread_t::io_thread_t (ctx_t *ctx_, uint32_t tid_) : object_t (ctx_, tid_) { poller = new (std::nothrow) poller_t; alloc_assert (poller); mailbox_handle = poller->add_fd (mailbox.get_fd (), this); poller->set_pollin (mailbox_handle); }
构造函数中把mailbox_t句柄加入poller中，让poller监听其读事件，所以，如果有命令发过来，poller会被唤醒，并调用io_thread_t的in_event：
void zmq::io_thread_t::in_event () { // TODO: Do we want to limit number of commands I/O thread can // process in a single go? command_t cmd; int rc = mailbox.recv (&cmd, 0); while (rc == 0 || errno == EINTR) {//如果读管道中有内容或者等待信号的时候被中断，将一直读取 if (rc == 0) cmd.destination->process_command (cmd); rc = mailbox.recv (&cmd, 0); } errno_assert (rc != 0 && errno == EAGAIN); }
可以看到，in_event使用了mailbox_t的接收命令的功能。接收到命令之后，调用destination处理命令的功能去处理命令。
上一节说过socket_base_t的每个实例都可以看成一个zmq线程，但是比较特殊，并没有使用poller，而是在使用到socket的下面几个方法的时候去检查是否有未处理的命令：
int zmq::socket_base_t::getsockopt (int option_, void *optval_,size_t *optvallen_) int zmq::socket_base_t::bind (const char *addr_) int zmq::socket_base_t::connect (const char *addr_) int zmq::socket_base_t::term_endpoint (const char *addr_) int zmq::socket_base_t::send (msg_t *msg_, int flags_) int zmq::socket_base_t::recv (msg_t *msg_, int flags_) void zmq::socket_base_t::in_event ()//这个函数只有在销毁socke的时候会被用到，在后面讲zmq_close的时候会说到
检查的手段就是调用process_commands方法：
int zmq::socket_base_t::process_commands (int timeout_, bool throttle_) { int rc; command_t cmd; if (timeout_ != 0) { // If we are asked to wait, simply ask mailbox to wait. rc = mailbox.recv (&cmd, timeout_); } else { some code rc = mailbox.recv (&cmd, 0); } // Process all available commands. while (rc == 0) { cmd.destination->process_command (cmd); rc = mailbox.recv (&cmd, 0); } some code }
可见，最终都是使用mailbox_t的接收命令的功能。
这里有一个值得思考的问题，为什么socket_base_t实例对应的这个线程不使用poller呢？每次使用上面那些方法的时候去检查不是很麻烦吗？
有关mailbox_t的实现，在下节介绍
标签：
原文地址：http://www.cnblogs.com/zengzy/p/5132437.html
