在初次介绍
发布/订阅模式的时候,就已经抖出了这个包袱:如果订阅者的消费速度慢,则会造成发布者端队列堆积,怎么办?本篇即是针对可能出现的"蜗牛"般的订阅者而生。
通常的做法:
在发布端用靠谱的队列承接来不及被消费的信息,这样会增大发布端的压力。
在订阅端用靠谱的队列承接来不及消费的信息,压力转嫁给各订阅端。
与前面相似,在队列中设置阈值,溢出则不收录。
压迫订阅端,当发布端发现订阅端过慢,给予惩罚性质的断开连接。
虽然上述4种方案都还算经典,不过总有欠妥之处。最好的方法莫过于让订阅者明白自个儿能力不足,作出主动性措施,比如:暂时性断开,优化了再来~
至于如何让订阅端检测是否消费能力不足,只需在订阅端设置一个有阈值的队列缓存数据,发布端给所有数据打上标号,如果订阅端读到“断号”,即可认定有数据超出阈值被舍弃了,那么,嘿嘿,主动整改吧。
虽然说中断订阅连接不是太理想,不过,总比不可知,不可控的一味流转数据要可靠的多,只少,都还掌控在自己手中~
//
// Suicidal Snail
//
#include "czmq.h"
// ---------------------------------------------------------------------
// This is our subscriber
// It connects to the publisher and subscribes to everything. It
// sleeps for a short time between messages to simulate doing too
// much work. If a message is more than 1 second late, it croaks.
#define MAX_ALLOWED_DELAY 1000 // msecs
static void
subscriber (void *args, zctx_t *ctx, void *pipe)
{
// Subscribe to everything
void *subscriber = zsocket_new (ctx, ZMQ_SUB);
zsocket_connect (subscriber, "tcp://localhost:5556");
// Get and process messages
while (1) {
char *string = zstr_recv (subscriber);
int64_t clock;
int terms = sscanf (string, "%" PRId64, &clock);
assert (terms == 1);
free (string);
// Suicide snail logic
if (zclock_time () - clock > MAX_ALLOWED_DELAY) {
fprintf (stderr, "E: subscriber cannot keep up, aborting\n");
break;
}
// Work for 1 msec plus some random additional time
zclock_sleep (1 + randof (2));
}
zstr_send (pipe, "gone and died");
}
// ---------------------------------------------------------------------
// This is our server task
// It publishes a time-stamped message to its pub socket every 1ms.
static void
publisher (void *args, zctx_t *ctx, void *pipe)
{
// Prepare publisher
void *publisher = zsocket_new (ctx, ZMQ_PUB);
zsocket_bind (publisher, "tcp://*:5556");
while (1) {
// Send current clock (msecs) to subscribers
char string [20];
sprintf (string, "%" PRId64, zclock_time ());
zstr_send (publisher, string);
char *signal = zstr_recv_nowait (pipe);
if (signal) {
free (signal);
break;
}
zclock_sleep (1); // 1msec wait
}
}
// This main thread simply starts a client, and a server, and then
// waits for the client to signal it's died.
//
int main (void)
{
zctx_t *ctx = zctx_new ();
void *pubpipe = zthread_fork (ctx, publisher, NULL);
void *subpipe = zthread_fork (ctx, subscriber, NULL);
free (zstr_recv (subpipe));
zstr_send (pubpipe, "break");
zclock_sleep (100);
zctx_destroy (&ctx);
return 0;
}
注意:
这里并未用到数据编号,而是打上了时间戳来判定数据是否延迟到不可接受的地步了,如何自我判定,方法万千,择优即可。
(未完待续)
分享到:
相关推荐
zeromq-2.1.7.tar.gz 的早期的一个版本,本人已安装成功 放心使用
zeromq-4.0.3.tar.gz zeromq-4.0.3.tar.gz zeromq-4.0.3.tar.gz
zeromq-4.3.2.tar.gz,可在linux下编译安装,能够使用zeromq进行sorket开发,多线程,提升性能,效率,可以配合msgpack进行使用,是个好的扩展插件
zeromq-4.1.3.tar.gz,最新的zeromq的开发工具包,希望对开发者有用
官方离线安装包,测试可用。使用rpm -ivh [rpm完整包名] 进行安装
zeromq-4.1.8.tar.gz 有问题请联系
zeromq-2.1.9.tar.gz 这是zeromq linux 官方原版 请放心下载
zeromq-4.2.3.tar.gz 一个稳定的版本,可以安装使用!
0MQ version 4.3.4 stable, released on 2021/01/17
storm搭建所需资源
在官网下载zeromq太慢了,网速极不稳定,特意下载放在这里供大家下载,当然象征性地赚点 资源分
ZeroMQ是一个网络通讯库,其主要用来为分布式应用程序开发提供进程间通信(此处的进程既可以是同一台机器上的两个进程也可以是不同机器上的两个进程)。ZeroMQ的特点在于灵活的通信手段和丰富的连接模型,并且它可以...
ZMQ(以下ZeroMQ简称ZMQ)是一个简单好用的传输层,像框架一样的一个socket library,他使得Socket编程更加简单、简洁和性能更高。是一个消息处理队列库,可在多个线程、内核和主机盒之间弹性伸缩。
ZeroMQ是一个网络通讯库,其主要用来为分布式应用程序开发提供进程间通信(此处的进程既可以是同一台机器上的两个进程也可以是不同机器上的两个进程)。ZeroMQ的特点在于灵活的通信手段和丰富的连接模型,并且它可以...
zeroMQ 4.2.3版本 zeromq-4.2.3.tar.gz 欢迎关注我的CSDN博客:https://mp.csdn.net/console/home 免积分下载
0MQ version 4.3.4 stable, released on 2021/01/17
ZeroMQ(也说明 ØMQ,0MQ 或 ZMQ)是一个高性能的异步消息库,旨在使用分布式或并行应用程序。它提供了一个消息队列,但 不同于面向消息的中间件,一个 ZeroMQ 系统可以在没有专用运行消息代理。jzmq安装包,本人也...
zeromq-4.2.5.tar.gz
VS2015 在Widows 10 上编译的 ZeroMQ 4.3.2,JZMQ 3.1 CZMQ 4.2,可以在 JDK 1.8 下运行。DLL 都是 64位,包含了编译及运行相关信息。分享一下,也给自己留个备份