Linux下使用ZMQ实践之监控事件

本文主要是介绍Linux下使用ZMQ实践之监控事件，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

1. 前言

继续《Linux下使用ZMQ实践“生产者-消费者”模型》文章之后进一步思考：
ZeroMQ通过隐藏了基础的socket操作，达到调用简明易懂的层次；
那么，如果某些场景下，又需要考虑到连接状态的维护，应该如何操作？
ZeroMQ给出的解决方案就是zmq_socket_monitor

2. 相关知识

支持监控的事件：

事件	描述
ZMQ_EVENT_CONNECTED	socket已被成功连接
ZMQ_EVENT_CONNECT_DELAYED	连接动作被挂起
ZMQ_EVENT_CONNECT_RETRIED	连接失败，正在重试
ZMQ_EVENT_LISTENING	监听成功
ZMQ_EVENT_BIND_FAILED	绑定失败
ZMQ_EVENT_ACCEPTED	接受新连接
ZMQ_EVENT_ACCEPT_FAILED	接受新连接失败
ZMQ_EVENT_CLOSED	socket关闭（主动关闭）
ZMQ_EVENT_CLOSE_FAILED	socket关闭失败
ZMQ_EVENT_DISCONNECTED	连接意外关闭（被关闭）
ZMQ_EVENT_MONITOR_STOPPED	监控的socket消亡

使用思路：将要监听的sock跟monitor关联，然后创建一个额外的ZMQ_PAIR，通过pair来获取sock上的事件。

3.场景举例

根据之前的“生产者-消费者”模型的一个改进：

已知固定的消费者个数，如4个；
生产者等待4个消费者全部启动后，才开始发送消息；
生产者发送退出消息，等待消费者断开连接后才最后退出；

在之前一对多的 Push-Pull 模式下，如果没有消费者连接，则生产者数据发送会一直阻塞，但如果有至少一个连接成功，则生产者进入发送数据阶段；在改进场景中，需求所有消费者就绪后，生产者才正式开始发送数据，达到一个理想的均衡状态。

这样，我们就依赖monitor机制的实现，监听消费者的Push套件，额外增加一个监听器monitor：

#define ADDR "tcp://127.0.0.1:555"
#define MONITOR "inproc://monitor-server"...void *sock = zmq_socket(ctx, ZMQ_PUSH);void *mon  = zmq_socket(ctx, ZMQ_PAIR);...zmq_bind(sock, ADDR);zmq_socket_monitor(sock, MONITOR, ZMQ_EVENT_ALL);zmq_connect(mon, MONITOR);...

下来，我们通过monitor等待4个消费者的连接事件，成功后才发送数据；
发送数据完成后，我们通过发送“Quit”报文来通知消费者退出进程；
完整的生产者代码如下：

void test_producer(void *ctx, int times)
{int ix = 0, cnt = 0, id = 0, event = 0;char request[1024];void *sock = zmq_socket(ctx, ZMQ_PUSH);void *mon  = zmq_socket(ctx, ZMQ_PAIR);s_set_id_num(sock, id);zmq_bind(sock, ADDR);zmq_socket_monitor(sock, MONITOR, ZMQ_EVENT_ALL);zmq_connect(mon, MONITOR);LOGN("Producer %d setup\n", id);for (cnt = 0; cnt < 4;) {event = get_monitor_event(mon, NULL, NULL);if (event == ZMQ_EVENT_ACCEPTED) {LOGN("Producer accepted\n");cnt++;}}LOGN("Producer %d start\n", id);for (ix = 0; ix < times; ix++) {snprintf(request, sizeof(request), "Data-%03d-%03d", id, ix);s_send(sock, request);LOGN("Producer %d send: %s\n", id, request);usleep(100 * 1000);}for (cnt = 0; cnt < 4;) {s_send(sock, "Quit"); // 通知一个消费者，退出一个消费者event = get_monitor_event(mon, NULL, NULL);if (event == ZMQ_EVENT_DISCONNECTED) {cnt++;}}LOGN("Producer %d stop\n", id);zmq_close(sock);
}

获取监听事件的接口为，get_monitor_event，该函数从ZeroMQ帮助手册摘抄下来：

static int get_monitor_event (void *monitor, int *value, char **address)
{   // First frame in message contains event number and valuezmq_msg_t msg;zmq_msg_init (&msg);if (zmq_msg_recv (&msg, monitor, 0) == -1)return -1; // Interrupted, presumablyassert (zmq_msg_more (&msg));uint8_t *data = (uint8_t *) zmq_msg_data (&msg);uint16_t event = *(uint16_t *) (data);if (value) *value = *(uint32_t *) (data + 2);// Second frame in message contains event addresszmq_msg_init (&msg);if (zmq_msg_recv (&msg, monitor, 0) == -1)return -1; // Interrupted, presumablyassert (!zmq_msg_more (&msg));if (address) {uint8_t *data = (uint8_t *) zmq_msg_data (&msg);size_t size = zmq_msg_size (&msg); *address = (char *) malloc (size + 1);memcpy (*address, data, size);(*address)[size] = 0;}return event;
}

然后消费者的实现，跟先前的例子差不多，就多了一个退出的判断：

int test_consumer(void *ctx, int id)
{int cnt = 0;char request[1024];void *sock = zmq_socket(ctx, ZMQ_PULL);s_set_id_num(sock, id);zmq_connect(sock, ADDR);LOGN("Consumer %d start\n", id);while (++cnt) {s_recv(sock, request);LOGN("Consumer %d recv: %s\n", id, request);usleep(300 * 1000);if (strcmp(request, "Quit") == 0) {break;}}LOGN("Consumer %d stop\n", id);zmq_close(sock);
}

最后，main函数功能，主要为fork，主进程做生产者，子进程做消费者；
同时，为了方便起见，省略了waitpid回收子进程的动作；


int main(int argc, char *argv[])
{int ix = 0;void *ctx = zmq_ctx_new();srandom(time(NULL));/* 1x producter vs 4x consumer */for (ix= 1; ix <= 4; ix++) {pid_t pid = fork();if (pid == 0) {test_consumer(ctx, ix);goto out;}}test_producer(ctx, atoi(argv[1]));// TODO waitpid
out:zmq_ctx_destroy(ctx);exit(EXIT_SUCCESS);
}

实际运行情况如下：

[ 1561228921.433 ]: Consumer 1 start
[ 1561228921.433 ]: Consumer 2 start
[ 1561228921.434 ]: Consumer 4 start
[ 1561228921.434 ]: Consumer 3 start
[ 1561228921.434 ]: Producer 0 setup
[ 1561228921.435 ]: Producer accepted
[ 1561228921.496 ]: Producer accepted
[ 1561228921.572 ]: Producer accepted
[ 1561228921.572 ]: Producer accepted
[ 1561228921.572 ]: Producer 0 start
[ 1561228921.572 ]: Producer 0 send: Data-000-000
[ 1561228921.574 ]: Consumer 3 recv: Data-000-000
[ 1561228921.673 ]: Producer 0 send: Data-000-001
[ 1561228921.774 ]: Producer 0 send: Data-000-002
[ 1561228921.775 ]: Consumer 2 recv: Data-000-002
[ 1561228921.876 ]: Producer 0 send: Data-000-003
[ 1561228921.877 ]: Consumer 1 recv: Data-000-003
[ 1561228921.978 ]: Producer 0 send: Data-000-004
[ 1561228921.978 ]: Consumer 4 recv: Data-000-004
[ 1561228922.079 ]: Producer 0 send: Data-000-005
[ 1561228922.081 ]: Consumer 3 recv: Data-000-005
[ 1561228922.183 ]: Producer 0 send: Data-000-006
[ 1561228922.284 ]: Producer 0 send: Data-000-007
[ 1561228922.285 ]: Consumer 2 recv: Data-000-007
[ 1561228922.386 ]: Producer 0 send: Data-000-008
[ 1561228922.387 ]: Consumer 1 recv: Data-000-008
[ 1561228922.488 ]: Producer 0 send: Data-000-009
[ 1561228922.488 ]: Consumer 4 recv: Data-000-009
[ 1561228922.590 ]: Consumer 3 recv: Quit
[ 1561228922.892 ]: Consumer 3 stop
[ 1561228922.894 ]: Consumer 2 recv: Quit
[ 1561228923.195 ]: Consumer 2 stop
[ 1561228923.196 ]: Consumer 1 recv: Quit
[ 1561228923.497 ]: Consumer 1 stop
[ 1561228923.499 ]: Consumer 4 recv: Quit
[ 1561228923.800 ]: Consumer 4 stop
[ 1561228923.802 ]: Producer 0 stop

可以看出，程序第一阶段，启动进程；第二阶段，发送数据，负载均衡；第三阶段，回收资源。