本文主要是介绍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
可以看出，程序第一阶段，启动进程；第二阶段，发送数据，负载均衡；第三阶段，回收资源。
4 结论
ZMQ监控事件的方法，提供了一种可选的扩展场景支持，实际使用可以放主线程处理，也可以放独立的子线程处理。
这篇关于Linux下使用ZMQ实践之监控事件的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！
