Method Process

phxqueue/consumer/consumer.cpp:566–778

}

comm::RetCode Consumer::Process(comm::proto::ConsumerContext &cc) {

    QLInfo("cc.sub_id %d cc.store_id %d cc.queue_id %d", cc.sub_id(), cc.store_id(), cc.queue_id());

    comm::RetCode ret;

    shared_ptr<const config::TopicConfig> topic_config;
    if (comm::RetCode::RET_OK != (ret = config::GlobalConfig::GetThreadInstance()->GetTopicConfigByTopicID(impl_->topic_id, topic_config))) {
        QLErr("GetTopicConfigByTopicID ret %d topic_id %d", comm::as_integer(ret), impl_->topic_id);
        return ret;
    }

    if (!topic_config->IsValidQueue(cc.queue_id(), -1, cc.sub_id())) {
        QLErr("IsValidQueue return false queue_id %d", cc.queue_id());
        return comm::RetCode::RET_ERR_RANGE_QUEUE;
    }

    shared_ptr<const config::proto::QueueInfo> queue_info;
    if (comm::RetCode::RET_OK != (ret = topic_config->GetQueueInfoByQueue(cc.queue_id(), queue_info))) {
        QLErr("GetQueueInfoByQueue ret %d queue_id %d", comm::as_integer(ret), cc.queue_id());
        return ret;
    }

    const int topic_id{topic_config->GetProto().topic().topic_id()};

    uint32_t maxlimit{topic_config->GetProto().topic().batch_limit()};
    uint32_t limit{topic_config->GetProto().topic().batch_limit()};

    const int sleep_us_per_get = queue_info->sleep_us_per_get();
    const int sleep_us_on_get_fail = queue_info->sleep_us_on_get_fail();
    const int sleep_us_on_get_no_item = queue_info->sleep_us_on_get_no_item();
    const int sleep_us_on_get_size_too_small = queue_info->sleep_us_on_get_size_too_small();


    const int queue_delay{queue_info->delay()};
    if (queue_delay < 0) {
        comm::ConsumerBP::GetThreadInstance()->OnGetQueueDelayFail(cc);
        QLErr("GetQueueDelay ret %d", queue_delay);
        return comm::RetCode::RET_ERR_RANGE;
    }

    auto start_loop_time = comm::utils::Time::GetSteadyClockMS();


    static double freq_quota = 0;
    static uint64_t last_fill_time = 0;

    static uint64_t last_topic_config_mod_time = 0;
    {
        auto new_topic_config_mod_time = topic_config->GetLastModTime();
        if (last_topic_config_mod_time != new_topic_config_mod_time) {
            last_topic_config_mod_time = new_topic_config_mod_time;
            if (freq_quota > 0) freq_quota = 0;
        }
    }

    while (true) {
        comm::ConsumerBP::GetThreadInstance()->OnLoop(cc);