第二个经典模式是单向数据分发，服务器推送更新到一组客户端。让我们看一个推送天气情况变化的例子，包含地区编码、温度、和相对湿度。我们会生成随机值来模拟真实气象站。
这是服务器代码，这个程序我们使用5556端口。
wuserver: Weather update server in C
[cpp]
//
// Weather update server
// Binds PUB socket to tcp://*:5556
// Publishes random weather updates
//
#include "zhelpers.h"
int main (void)
{
// Prepare our context and publisher
void *context = zmq_ctx_new ();
void *publisher = zmq_socket (context, ZMQ_PUB);
int rc = zmq_bind (publisher, "tcp://*:5556");
assert (rc == 0);
rc = zmq_bind (publisher, "ipc://weather.ipc");
assert (rc == 0);
// Initialize random number generator
srandom ((unsigned) time (NULL));
while (1) {
// Get values that will fool the boss
int zipcode, temperature, relhumidity;
zipcode = randof (100000);
temperature = randof (215) - 80;
relhumidity = randof (50) + 10;
// Send message to all subscribers
char update [20];
sprintf (update, "%05d %d %d", zipcode, temperature, relhumidity);
s_send (publisher, update);
}
zmq_close (publisher);
zmq_ctx_destroy (context);
return 0;
}
更新流既无开始也无结束，像一个永不结束的天气预报。
图 4 – 发布-订阅
这是客户端程序，监听更新流并捕获符合特定地区编码的所有消息，默认为纽约市因为那是个冒险的好地方：
wuclient: Weather update client in C
[cpp]
//
// Weather update client
// Connects SUB socket to tcp://localhost:5556
// Collects weather updates and finds avg temp in zipcode
//
#include "zhelpers.h"
int main (int argc, char *argv [])
{
void *context = zmq_ctx_new ();
// Socket to talk to server
printf ("Collecting updates from weather server…\n");
void *subscriber = zmq_socket (context, ZMQ_SUB);
int rc = zmq_connect (subscriber, "tcp://localhost:5556");
assert (rc == 0);
// Subscribe to zipcode, default is NYC, 10001
char *filter = (argc > 1) argv [1] : "10001 ";
rc = zmq_setsockopt (subscriber, ZMQ_SUBSCRIBE, filter, strlen (filter));
assert (rc == 0);
// Process 100 updates
int update_nbr;
long total_temp = 0;
for (update_nbr = 0; update_nbr < 100; update_nbr++) {
char *string = s_recv (subscriber);
int zipcode, temperature, relhumidity;
sscanf (string, "%d %d %d",
&zipcode, &temperature, &relhumidity);
total_temp += temperature;
free (string);
}
printf ("Average temperature for zipcode '%s' was %dF\n",
filter, (int) (total_temp / update_nbr));
zmq_close (subscriber);
zmq_ctx_destroy (context);
return 0;
}
注意当你使用一个订阅套接字时你必须使用zmq_setsockopt()和SUBSCRIBE设置一个订阅，就像这段代码中那样。如果你不设置任何订阅，就得不到任何消息。这是初学者的常见错误。订阅者可以设置很多订阅，会合并到一起。就是说，如果一个更新匹配任意一个订阅，订阅者都会接收。订阅者也可以取消特定的订阅。一个订阅通常是一个可打印字符串，但也不是必须的。参考zmq_setsockopt()看这是怎么工作的。
发布订阅套接字对是异步的。客户端在循环中（或单次如果有必要）做zmq_msg_recv()。尝试发送消息到订阅套接字将导致错误。同样的，服务按所需频率做zmq_msg_send()，但绝不能对发布套接字做zmq_msg_recv()。
理论上， MQ套接字不在乎哪一端来连接哪一端来绑定。但是在实践中会有未公开的差异，待会我会提及。现在，绑定发布并连接订阅，除非你的网络设计导致这无法实现。
还有一个关于发布订阅套接字的重要事项：你无法精确知道订阅者什么时候开始获取消息。即使你先启动一个订阅者，过一会再启动发布者，订阅者总是会错过发布者发送的第一条消息。这是因为订阅者连接到发布者时（占用了短暂但非零的时间），发布者可能已经将消息发送出去了。
这种“迟钝加入者”症状击中了很多人、很多次，我们需要详细解释一下。记住 MQ是异步I/O的，也就是在后台。比如说你有两个节点按这个顺序这么做：
订阅者连接到一个