登录
当前位置： 首页 > 知识库问答 >
问题：

librdkafka使用者未从代理接收消息

龙枫
2023-03-14

我已经用C语言编写了kafka消费者和生产者，使用的库是Library dkafka库。kafka代理版本是kafka_2.12-2.3.0。生产者正在成功生成消息，dr_msg_cb函数确认成功传递。但是，消费者没有收到来自代理的消息。有人能帮助进一步调试吗？

我可以看到，从消费者到代理的TCP连接已经建立。但TCPdump显示代理并没有向消费者发送任何数据。我在消费者代码上启用了调试，下面是消息。

[2019 Nov  8 19:18:09.458553135:155:E:logger:1741] TID 05 : [LOG_TRACE]:RDKAFKA-7-SSL: rdkafka#consumer-2: [thrd:app]: Loading CA certificate(s) from file /mnt/ifc/cfg/d
ata/securedata/clientcerts/kafka/ApicCa.crt
[2019 Nov  8 19:18:09.458880860:156:E:logger:1741] TID 05 : [LOG_TRACE]:RDKAFKA-7-SSL: rdkafka#consumer-2: [thrd:app]: Loading certificate from file /mnt/ifc/cfg/data/se
curedata/clientcerts/kafka/KafkaClient.crt
[2019 Nov  8 19:18:09.459151178:157:E:logger:1741] TID 05 : [LOG_TRACE]:RDKAFKA-7-SSL: rdkafka#consumer-2: [thrd:app]: Loading private key file from /mnt/ifc/cfg/data/se
curedata/clientcerts/kafka/KafkaClient8.key
[2019 Nov  8 19:18:09.459583515:158:E:logger:1741] TID 06 : [LOG_TRACE]:RDKAFKA-7-BRKMAIN: rdkafka#consumer-2: [thrd::0/internal]: :0/internal: Enter main broker thread
[2019 Nov  8 19:18:09.459589163:159:E:logger:1741] TID 06 : [LOG_TRACE]:RDKAFKA-7-STATE: rdkafka#consumer-2: [thrd::0/internal]: :0/internal: Broker changed state INIT -
> UP
[2019 Nov  8 19:18:09.459593374:160:E:logger:1741] TID 06 : [LOG_TRACE]:RDKAFKA-7-BROADCAST: rdkafka#consumer-2: [thrd::0/internal]: Broadcasting state change
[2019 Nov  8 19:18:09.459608395:161:E:logger:1741] TID 07 : [LOG_TRACE]:RDKAFKA-7-BROADCAST: rdkafka#consumer-2: [thrd:main]: Broadcasting state change
[2019 Nov  8 19:18:09.459708091:162:E:logger:1741] TID 05 : [LOG_TRACE]:RDKAFKA-7-BROKER: rdkafka#consumer-2: [thrd:app]: ssl://10.0.0.1:9092/bootstrap: Added new broker with NodeId -1
[2019 Nov  8 19:18:09.459718029:163:E:logger:1741] TID 08 : [LOG_TRACE]:RDKAFKA-7-BRKMAIN: rdkafka#consumer-2: [thrd:ssl://10.0.0.1:9092/bootstrap]: ssl://10.0.0.1:9092/bootstrap: Enter main broker thread
[2019 Nov  8 19:18:09.459723538:164:E:logger:1741] TID 08 : [LOG_TRACE]:RDKAFKA-7-CONNECT: rdkafka#consumer-2: [thrd:ssl://10.0.0.1:9092/bootstrap]: ssl://10.0.0.1:9092/bootstrap: broker in state INIT connecting
[2019 Nov  8 19:18:09.459918518:165:E:logger:1741] TID 08 : [LOG_TRACE]:RDKAFKA-7-CONNECT: rdkafka#consumer-2: [thrd:ssl://10.0.0.1:9092/bootstrap]: ssl://10.0.0.1:9092/bootstrap: Connecting to ipv4#10.0.0.1:9092 (ssl) with socket 34
[2019 Nov  8 19:18:09.460017515:166:E:logger:1741] TID 08 : [LOG_TRACE]:RDKAFKA-7-STATE: rdkafka#consumer-2: [thrd:ssl://10.0.0.1:9092/bootstrap]: ssl://10.0.0.1:9092/bootstrap: Broker changed state INIT -> CONNECT
[2019 Nov  8 19:18:09.460021977:167:E:logger:1741] TID 08 : [LOG_TRACE]:RDKAFKA-7-BROADCAST: rdkafka#consumer-2: [thrd:ssl://10.0.0.1:9092/bootstrap]: Broadcasting state change
[2019 Nov  8 19:18:09.460228677:168:E:logger:1741] TID 08 : [LOG_TRACE]:RDKAFKA-7-CONNECT: rdkafka#consumer-2: [thrd:ssl://10.0.0.1:9092/bootstrap]: ssl://10.0.0.1:9092/bootstrap: Connected to ipv4#10.0.0.1:9092
[2019 Nov  8 19:18:09.790145695:169:E:logger:1741] TID 08 : [LOG_TRACE]:RDKAFKA-7-SSLVERIFY: rdkafka#consumer-2: [thrd:ssl://10.0.0.1:9092/bootstrap]: ssl://10.0.0.1:9092/bootstrap: Broker SSL certificate verified
[2019 Nov  8 19:18:09.790151895:170:E:logger:1741] TID 08 : [LOG_TRACE]:RDKAFKA-7-CONNECTED: rdkafka#consumer-2: [thrd:ssl://10.0.0.1:9092/bootstrap]: ssl://10.0.0.1:9092/bootstrap: Connected (#1)
[2019 Nov  8 19:18:09.790168266:171:E:logger:1741] TID 08 : [LOG_TRACE]:RDKAFKA-7-FEATURE: rdkafka#consumer-2: [thrd:ssl://10.0.0.1:9092/bootstrap]: ssl://10.0.0.1:9092/bootstrap: Updated enabled protocol features +ApiVersion to ApiVersion
[2019 Nov  8 19:18:09.790172810:172:E:logger:1741] TID 08 : [LOG_TRACE]:RDKAFKA-7-STATE: rdkafka#consumer-2: [thrd:ssl://10.0.0.1:9092/bootstrap]: ssl://10.0.0.1:9092/bootstrap: Broker changed state CONNECT -> APIVERSION_QUERY
[2019 Nov  8 19:18:09.790176880:173:E:logger:1741] TID 08 : [LOG_TRACE]:RDKAFKA-7-BROADCAST: rdkafka#consumer-2: [thrd:ssl://10.0.0.1:9092/bootstrap]: Broadcasting state change
[2019 Nov  8 19:18:09.790888559:174:E:logger:1741] TID 08 : [LOG_TRACE]:RDKAFKA-7-FEATURE: rdkafka#consumer-2: [thrd:ssl://10.0.0.1:9092/bootstrap]: ssl://10.0.0.1:9092/bootstrap: Updated enabled protocol features to MsgVer1,ApiVersion,BrokerBalancedConsumer,ThrottleTime,Sasl,SaslHandshake,BrokerGroupCoordinator,LZ4,OffsetTime,MsgVer2
[2019 Nov  8 19:18:09.790893525:175:E:logger:1741] TID 08 : [LOG_TRACE]:RDKAFKA-7-STATE: rdkafka#consumer-2: [thrd:ssl://10.0.0.1:9092/bootstrap]: ssl://10.0.0.1:9092/bootstrap: Broker changed state APIVERSION_QUERY -> UP
[2019 Nov  8 19:18:09.790897645:176:E:logger:1741] TID 08 : [LOG_TRACE]:RDKAFKA-7-BROADCAST: rdkafka#consumer-2: [thrd:ssl://10.0.0.1:9092/bootstrap]: Broadcasting state change
[2019 Nov  8 19:18:09.791643149:177:E:logger:1741] TID 07 : [LOG_TRACE]:RDKAFKA-7-CLUSTERID: rdkafka#consumer-2: [thrd:main]: ssl://10.0.0.1:9092/bootstrap: ClusterId update "" -> "r7Us-jYGQRq34re8owKyJA"
[2019 Nov  8 19:18:09.791654890:178:E:logger:1741] TID 08 : [LOG_TRACE]:RDKAFKA-7-UPDATE: rdkafka#consumer-2: [thrd:ssl://10.0.0.1:9092/bootstrap]: ssl://10.0.0.1:9092/bootstrap: NodeId changed from -1 to 1
[2019 Nov  8 19:18:09.791663562:179:E:logger:1741] TID 08 : [LOG_TRACE]:RDKAFKA-7-UPDATE: rdkafka#consumer-2: [thrd:ssl://10.0.0.1:9092/bootstrap]: ssl://10.0.0.1:9092/1: Name changed from ssl://10.0.0.1:9092/bootstrap to ssl://10.0.0.1:9092/1
[2019 Nov  8 19:18:09.791668295:180:E:logger:1741] TID 08 : [LOG_TRACE]:RDKAFKA-7-LEADER: rdkafka#consumer-2: [thrd:ssl://10.0.0.1:9092/bootstrap]: ssl://10.0.0.1:9092/1: Mapped 0 partition(s) to broker
[2019 Nov  8 19:18:09.791671709:181:E:logger:1741] TID 08 : [LOG_TRACE]:RDKAFKA-7-STATE: rdkafka#consumer-2: [thrd:ssl://10.0.0.1:9092/bootstrap]: ssl://10.0.0.1:9092/1: Broker changed state UP -> UPDATE
[2019 Nov  8 19:18:09.791675360:182:E:logger:1741] TID 08 : [LOG_TRACE]:RDKAFKA-7-BROADCAST: rdkafka#consumer-2: [thrd:ssl://10.0.0.1:9092/bootstrap]: Broadcasting state change
[2019 Nov  8 19:18:09.791692544:183:E:logger:1741] TID 08 : [LOG_TRACE]:RDKAFKA-7-STATE: rdkafka#consumer-2: [thrd:ssl://10.0.0.1:9092/bootstrap]: ssl://10.0.0.1:9092/1: Broker changed state UPDATE -> UP
[2019 Nov  8 19:18:09.791696027:184:E:logger:1741] TID 08 : [LOG_TRACE]:RDKAFKA-7-BROADCAST: rdkafka#consumer-2: [thrd:ssl://10.0.0.1:9092/bootstrap]: Broadcasting state change

消费者代码如下所示。

static void
msg_consume(rd_kafka_message_t *rkmessage, void *opaque)
{
    if (rkmessage == NULL) {
        return;
    }

    if (rkmessage->err) {
        if (rkmessage->err == RD_KAFKA_RESP_ERR__PARTITION_EOF) {
            DEBUG_PRINT(DBG_TRACE,
                        "%% Consumer reached end of %s [%"PRId32"] "
                        "message queue at offset %"PRId64"\n",
                         (rkmessage->rkt) ? 
                          rd_kafka_topic_name(rkmessage->rkt) : "NULL",
                          rkmessage->partition, rkmessage->offset);
            return;
        }

        if (rkmessage->err == RD_KAFKA_RESP_ERR__UNKNOWN_PARTITION ||
            rkmessage->err == RD_KAFKA_RESP_ERR__UNKNOWN_TOPIC) {
            return;
        }

        return;
    }

    if (rkmessage->key_len) {
        DEBUG_PRINT(DBG_TRACE, "Key: %.*s\n", (int)rkmessage->key_len, 
                    (char *)rkmessage->key);
    }

    DEBUG_PRINT(DBG_TRACE, "%.*s\n", (int)rkmessage->len, 
                (char *)rkmessage->payload);
}

syserr_t
kafka_consumer_create()
{
    rd_kafka_topic_conf_t *consTopicCfg;
    rd_kafka_conf_t *conf = NULL;
    rd_kafka_t *rk = NULL;
    char errstr[512];
    rd_kafka_resp_err_t errCode;

    conf = rd_kafka_conf_new();

    if (!conf) {
        return ~SUCCESS;
    }

    rd_kafka_conf_set_log_cb(conf, logger);

    if (rd_kafka_conf_set(conf, "debug", 
                          "generic,broker,topic,security,msg,fetch",
                          errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
    }

    if (rd_kafka_conf_set(conf, "bootstrap.servers", broker_list[0],
                          errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
        return ~SUCCESS;
    }

    if (rd_kafka_conf_set(conf, "security.protocol", "SSL",
                          errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
        return ~SUCCESS;
    }

    if (rd_kafka_conf_set(conf, "ssl.certificate.location", kafka_clnt_cert,
                          errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
        return ~SUCCESS;
    }

    if (rd_kafka_conf_set(conf, "ssl.key.location", kafka_clnt_key,
                          errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
        return ~SUCCESS;
    }

    if (rd_kafka_conf_set(conf, "ssl.ca.location", kafka_apic_cert,
                          errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
        return ~SUCCESS;
    }

    if (rd_kafka_conf_set(conf, "auto.commit.enable", "true",
                          errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
        return ~SUCCESS;
    }

    if (rd_kafka_conf_set(conf, "auto.commit.interval.ms", "500",
                          errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
        return ~SUCCESS;
    }

    if (rd_kafka_conf_set(conf, "group.id", "consumerGroup",
                          errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
        return ~SUCCESS;
    }

    consTopicCfg = rd_kafka_topic_conf_new();

    if (RD_KAFKA_CONF_OK != rd_kafka_topic_conf_set(consTopicCfg, 
                                                    "auto.offset.reset",
                                                    "latest" ,errstr,
                                                    sizeof(errstr))) {
        return ~SUCCESS;
    }

    rd_kafka_conf_set_default_topic_conf(conf, consT opicCfg);
    rk = rd_kafka_new(RD_KAFKA_CONSUMER, conf, errstr, sizeof(errstr));
    if (!rk) {
        return ~SUCCESS;
    }

//    conf = NULL; // Disown conf as rd_kafka_new() has ownership now.

    const char *ep_topic="eprecords";
    rd_kafka_topic_partition_list_t *tp_list = 
                            rd_kafka_topic_partition_list_new(1);
    rd_kafka_topic_partition_t* tpObj = 
                            rd_kafka_topic_partition_list_add(tp_list,
                                                              ep_topic, 
                                                      RD_KAFKA_PARTITION_UA);

    if (NULL == tpObj) {
        return ~SUCCESS;
    }

    errCode = rd_kafka_subscribe(rk, tp_list);
    if (errCode  != RD_KAFKA_RESP_ERR_NO_ERROR) {
        return ~SUCCESS;
    }

    rd_kafka_topic_partition_list_destroy(tp_list);

    while(1) {
        rd_kafka_message_t *msg = rd_kafka_consumer_poll(rk, 1000);
        if (msg != NULL) {
            if (msg->err == RD_KAFKA_RESP_ERR_NO_ERROR) {
                msg_consume(msg, NULL);
            }
            rd_kafka_message_destroy(msg);
        }
        rd_kafka_poll(rk,0);
    }
}

我希望在producer定期发布数据时调用msg\u consume（）。我不确定下面的日志消息是否是问题的根源。

“[2019年11月8日19:18:09.791668295:180:E:logger:1741]TID 08:[日志跟踪]：RDKAFKA-7-领导者：RDKAFKA#消费者-2:[第三方：ssl://10.0.0.1:9092/bootstrap]: ssl://10.0.0.1:9092/1:已将0个分区映射到代理”

共有1个答案

江航
2023-03-14

在我换了一种不同的消费模式后，这个程序开始工作。代码如下。

static void logger (const rd_kafka_t *rk, int level,
                    const char *fac, const char *buf)
{
    PRINT_ERR("RDKAFKA-%i-%s: %s: %s", level, fac, rk ? rd_kafka_name(rk) : NULL, buf);
}

rd_kafka_t *
kafka_consumer_create()
{
    rd_kafka_conf_t *conf = NULL;
    rd_kafka_t *rk = NULL;
    char errstr[512];
    rd_kafka_resp_err_t errCode;

    conf = rd_kafka_conf_new();

    if (!conf) {
        return ~SUCCESS;
    }

    rd_kafka_conf_set_log_cb(conf, logger);

    if (rd_kafka_conf_set(conf, "enable.partition.eof", "true",
                          errstr, sizeof(errstr) != RD_KAFKA_CONF_OK)) {
        PRINT_ERR("enable partition eof failed %s", errstr);
        return ~SUCCESS;
    }

    if (rd_kafka_conf_set(conf, "client.id", "kafka-python-Python 2.7.16",
                          errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
        PRINT_ERR("DR_MSG : failed to set client id %s", errstr);
        return ~SUCCESS;
    }

    if (rd_kafka_conf_set(conf, "debug", "generic,broker,topic,security,msg,fetch",
                          errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
        return ~SUCCESS;
    }

    if (rd_kafka_conf_set(conf, "enable.auto.commit","true",
                          errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
        return ~SUCCESS;
    }

    if (rd_kafka_conf_set(conf, "bootstrap.servers", "192.168.1.1",
                          errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
        return ~SUCCESS;
    }

    if (rd_kafka_conf_set(conf, "security.protocol", "SSL",
                          errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
        return ~SUCCESS;
    }

    if (rd_kafka_conf_set(conf, "ssl.certificate.location", "/root/kafka/KafkaClient.crt",
                          errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
        return ~SUCCESS;
    }

    if (rd_kafka_conf_set(conf, "ssl.key.location", "/root/kafka/KafkaClient8.key",
                          errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
        return ~SUCCESS;
    }

    if (rd_kafka_conf_set(conf, "ssl.ca.location", "/root/kafka/ApicCa.crt",
                          errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
        return ~SUCCESS;
    }

    if (rd_kafka_conf_set(conf, "group.id", "test-consumer-group",
                          errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
        return ~SUCCESS;
    }

    if (rd_kafka_conf_set(conf, "auto.offset.reset", "latest",
                          errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
        return ~SUCCESS;
    }

    rk = rd_kafka_new(RD_KAFKA_CONSUMER, conf, errstr, sizeof(errstr));
    if (!rk) {
        return ~SUCCESS;
    }

    conf = NULL; // Disown conf as rd_kafka_new() has ownership now.

    return rk;
}

static void
msg_consume(rd_kafka_message_t *rkmessage, void *opaque)
{
    if (rkmessage == NULL) {
        PRINT_ERR("Aentp rkmessage error");
        return;
    }

    if (rkmessage->err) {
        if (rkmessage->err == RD_KAFKA_RESP_ERR__PARTITION_EOF) {
            PRINT_ERR("AT2");
            DEBUG_PRINT(DBG_AENTP_TRACE,
                        "%% Consumer reached end of %s [%"PRId32"] "
                        "message queue at offset %"PRId64"\n",
                        (rkmessage->rkt) ? rd_kafka_topic_name(rkmessage->rkt) : "NULL",
                        rkmessage->partition, rkmessage->offset);
            return;
        }

        if (rkmessage->err == RD_KAFKA_RESP_ERR__UNKNOWN_PARTITION ||
            rkmessage->err == RD_KAFKA_RESP_ERR__UNKNOWN_TOPIC) {
            PRINT_ERR("Unknown partition or unknow topic");
            return;
        }

        return;
    }

    if (rkmessage->key_len) {
        DEBUG_PRINT("Key: %.*s\n", (int)rkmessage->key_len, (char *)rkmessage->key);
    }

    DEBUG_PRINT("%.*s\n", (int)rkmessage->len, (char *)rkmessage->payload);
}

void *
kafka_consumer_thread(void *ptr)
{
    rd_kafka_t *rk = kafka_consumer_create();
    rd_kafka_topic_t *rkt = NULL;
    rd_kafka_topic_conf_t *topic_conf;
    rd_kafka_resp_err_t err;
    char errstr[512];

    if (rk == NULL) {        
        return NULL;
    }

    DEBUG_PRINT("Starting Kafka consumer thread");
    topic_conf = rd_kafka_topic_conf_new();
    if (RD_KAFKA_CONF_OK != rd_kafka_topic_conf_set(topic_conf, "auto.offset.reset",
                            "earliest" ,errstr, sizeof(errstr))) {
        AENTP_PRINT_ERR("rd_kafka_topic_conf_set() failed with error: %s\n", errstr);
        return NULL;
    }

    rkt = rd_kafka_topic_new(rk, "records", topic_conf);
    topic_conf = NULL; /* Now owned by topic */

    /* Start consuming */
    if (rd_kafka_consume_start(rkt, 0, RD_KAFKA_OFFSET_END) == -1){
        rd_kafka_resp_err_t err = rd_kafka_last_error();
        PRINT_ERR("%% Failed to start consuming: %s\n", rd_kafka_err2str(err));
        if (err == RD_KAFKA_RESP_ERR__INVALID_ARG) {
            APRINT_ERR("%% Broker based offset storage "
                       "requires a group.id, "
                       "add: -X group.id=yourGroup\n");
        }
        return NULL;
    }

    while(1) {
        rd_kafka_message_t *rkmessage;

        /* Poll for errors, etc. */
        rd_kafka_poll(rk, 0);

        /*
         * Consume single message.
         * See rdkafka_performance.c for high speed
         * consuming of messages.
         */
        rkmessage = rd_kafka_consume(rkt, 0, 1000);
        if (!rkmessage) /* timeout */
            continue;

        msg_consume(rkmessage, NULL);

        /* Return message to rdkafka */
        rd_kafka_message_destroy(rkmessage);

        /* XXX Do we need the seek??? */
        seek_offset = 0;
        if (seek_offset) {
            err = rd_kafka_seek(rkt, 0, seek_offset, 2000);
            if (err)
                AENTP_PRINT_ERR("Seek failed: %s\n", rd_kafka_err2str(err));
            else
                printf("Seeked to %"PRId64"\n", seek_offset);
            seek_offset = 0;
        }
        /* XXX Do we need the seek??? */
    }

    return NULL;
}

请务必注意，为什么前面的方法不起作用。

 类似资料：
  • 使用者未从ActiveMQ接收消息

    我们面临着ActiveMQ及其消费者的随机问题。我们观察到，很少有消费者不接收消息，即使他们连接到ActiveMQ队列。但在消费者重启后，它工作正常。 我们在ActiveMQ端有一个名为testQueue的队列。消费者正试图将消息从该队列中解列。为此，我们正在使用Spring的DefaultMessageListenerContainer。消息正在从ActiveMQ代理传递到使用者节点。从tcpd

  • kafka-python使用者未接收消息

    我在使用时遇到了困难，无法从开头或其他任何显式偏移量读取它。 为同一主题的使用者运行命令行工具，我确实看到带有选项的消息，否则它将挂起 我使用的是kafka-python 0.9.5，而代理运行的是Kafka8.2。不确定确切的问题是什么。 按照dpkp的建议设置_group_id=none_以模拟控制台使用者的行为。

  • Apache Kafka-消费者未从生产者接收消息

    我很感激你在这方面的帮助。 我正在构建一个ApacheKafka消费者，以订阅另一个已经运行的Kafka。现在，我的问题是，当我的制作人将消息推送到服务器时。。。我的消费者没有收到。。我在打印的日志中得到以下信息： 我不确定我是否遗漏了任何重要的配置。。。但是，我可以使用WireShark看到一些来自我的服务器的消息，但是我的消费者没有消费这些消息。。。。 我的代码是示例消费者示例的精确副本：ht

  • ActiveMQ未从生产者接收消息

    我有一个JMS生产者和一个消费者，代理是ActiveMQ，参考下面的代码： 寄件人代码 接收码 问题是 ActiveMQ 队列无法接收来自发送方的消息（请参阅屏幕截图）： 当我从 Web 控制台发送消息时，该消息在队列中收到，但来自创建者的消息不会进入队列。 另一个有趣的行为是（如队列接收器代码中所示，接收器在收到第一条消息后退出），同样，当我启动接收器时，它会收到相同的消息，并继续执行，直到我关

  • 活动的MQ使用者只接收替代消息

    要求：一个activemq队列应该在一台机器（比如machine1)中运行。该队列应该从machine1和另一台机器（比如machine2)中访问。监听器将只在machine1中运行。因此，我发送到machine1和machine2的所有消息都应该进入machine1中可用的队列，并且应该被同一台机器(machine1)中可用的监听器接收。 问题：我发送到machine1和machine2的所有消

  • 消息代理适合简单的消息发送者和接收者吗？

    我正在参与一个项目，该项目将首先构建一个简单的消息系统，该系统将接收消息，存储消息并将其路由到适当的部门。基本用例是： < li >用户在网站表单中写下一条消息或一个问题，并选择一个部门将消息发送给该部门 < li >根据用户的选择，消息被发送到相应部门的消息队列，状态为“未读”、“已读”等。(我们还没有确定所有的状态)。 < li >这些信息成为用户与网站互动的一部分，即，如果用户拨打客服电话，

相关问答

活动的MQ使用者只接收替代消息消息代理适合简单的消息发送者和接收者吗？JmsListener未从Activemq接收消息Kafka Java消费者从未收到任何消息Kafka消费者不接收消息

相关文章

BrowserMob代理+ Selenium：未收到任何HTTP响应方法接收者的歧义命令模式如何使发送者与接收者脱钩？C++：代理者模式您可以使用什么来使应用程序能够接收SMS消息？

相关阅读

RabbitMQ是如何收发消息的？User-Agent用户代理处理有异常消息RebalanceService消费者负载均衡过程消费者DefaultMQPushConsumer启动主要流程

相关工具

librdkafka蓝牙接收与发送数据猫国建设者大块头微信消息青青世界之侠者江湖

相关文档

Windows 连接管理器管理工具包 帮助文档滴滴企业管理接口文档Windows 连接管理器 帮助文档EnvoyProxy 智能代理中文参考文档基于 Node.js 的 HTTPS MITM 中间人代理的原理和实现
快捷导航： 新手教程 算法原理 架构设计 Java进阶 数据库进阶 大厂专栏 面试经验 编程笔记 编程问答 所有专题 文档资料 工具软件 电子书籍 小牛导航
在线工具： 房贷计算器 个税计算器 Linux命令查询 Json格式化 正则表达式 颜色转换 AES加解密 SHA1加密 MD5加密 毒鸡汤 字数统计 随机密码生成 进制转换 Base64编解码 励志句子
Copyright © 2019-2024 小牛知识库@xnip.cn. All Rights Reserved.