Method MakeHandleBuckets

phxqueue/consumer/consumer.cpp:171–232

}

comm::RetCode Consumer::MakeHandleBuckets() {
    comm::ConsumerConsumeBP::GetThreadInstance()->OnMakeHandleBucket(impl_->cc);

    for (int i{0}; i < impl_->opt.nhandler; ++i) {
        while(!impl_->handle_buckets[i].empty())impl_->handle_buckets[i].pop();
    }

    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;
    }

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

    int nbucket_used{0};
    map<uint64_t, int> uin2bucket_idx;
    int max_sz = -1;
    int max_handle_id = -1;
    uint64_t max_key = -1;
    for (int i{0}; i < impl_->items.size(); ++i) {
        auto key = impl_->items[i]->meta().uin();
        auto handle_id = impl_->items[i]->meta().handle_id();

        int bucket_idx = -1;
        bool update_uin2bucket_idx = false;
        if (key && !queue_info->handle_by_random_uin()) {
            auto &&it = uin2bucket_idx.find(key);
            if (uin2bucket_idx.end() != it) {
                bucket_idx = it->second;
            } else {
                update_uin2bucket_idx = true;
            }
        }
        if (-1 == bucket_idx) {
            if (nbucket_used >= impl_->opt.nhandler) {
                bucket_idx = comm::utils::OtherUtils::FastRand() % impl_->opt.nhandler;
            } else {
                bucket_idx = nbucket_used++;
            }
        }
        if (update_uin2bucket_idx) uin2bucket_idx[key] = bucket_idx;

        impl_->handle_buckets[bucket_idx].push(i);
        auto sz = impl_->handle_buckets[bucket_idx].size();
        if (-1 == max_sz || sz > max_sz) {
            max_sz = sz;
            max_handle_id = handle_id;
            max_key = key;
        }
    }