主教堂的尾部入口区则使用宽大的罗马拱劵构成外界形象,特地用来存这么的parition

  • topic-partition是kafka布满式的精彩,
    也是指向kafka举办生产或费用的蝇头单元;
  • 在此篇里大家开端介绍有关的数据结构
  • 情节如下:
    1. rd_kafka_topic_partition_t
    2. rd_kafka_topic_partition_list_t
    3. rd_kafka_toppar_s

想必大家都有会有这么的沉郁:亲人非常的少,做多几样菜吧,太费劲,做少几样吧,吃得又不开玩笑。其实整后生可畏锅砂锅粥,就能够很好化解那么些难题,丰硕又轻巧。

文-木又


先扯大器晚成扯潮汕砂锅粥由来。

     
 西开教堂——圣Louis最大的天主教会,加尔各答天主教徒所集聚的敬拜地方,始建于1913年,于今已有百年历史。作为滨江大道上的侧着重景色,西开教堂统摄全局、吸收接纳旅客,匆匆岁月间已改中年大家出境游与旅游的“圣地”。而教堂为什么称为“西开”?本身的出行价值浮今后何方?,一贯是大家一向不关心的细节。带着奇异与难点,我单独来到“圣堂”,体会与观望那座城墙的“灵宫”——西开大教堂。

rd_kafka_topic_partition_t
  • 所在文件: src/rdkafka.h
  • 概念了三个partition的连锁数据结构, 轻巧定义, 占位符
  • 定义:

typedef struct rd_kafka_topic_partition_s {
        char        *topic;             /**< Topic name */
        int32_t      partition;         /**< Partition */
    int64_t      offset;            /**< Offset */
        void        *metadata;          /**< Metadata */ // 主要是leader, replicas, isr等信息
        size_t       metadata_size;     /**< Metadata size */
        void        *opaque;            /**< Application opaque */
        rd_kafka_resp_err_t err;        /**< Error code, depending on use. */
        void       *_private;           /**< INTERNAL USE ONLY,
                                         *   INITIALIZE TO ZERO, DO NOT TOUCH */
} rd_kafka_topic_partition_t;

从前潮汕人靠海为生,出海嘛随船也就带点米啥的。

     
 秋色之下,作者徒步至此,光洁的街道烘托着铅色的基调,纷纷的落叶拨弄着空气的节拍。远观教堂,罗曼式的建筑风格散发出神秘的鼻息,隐深的大门引领着脚步的点子,拱形的门窗诉说着古老的奥密,前后错落的松木与出挑的树枝也在美化着教堂的构图,全体意况就好像Sargent笔下是颜色画,朦胧润色。走进院落后,作者早先细细体积那座建筑,开采教堂的入口区朝西南向,全体立面分三段式构图。其上层塔楼对称而立,塔上冠以穹隆,中度可达45米,表皮以北京蓝铜版为饰面,塔尖之上还停放了青铜十字架,色彩搭配淡青青古铜色,美轮美奂。其下,中层三角德州花嵌于双塔之间,檐口则动用暗灰券柱列举行装修,突显了建造的层落关系。而基本玻璃彩窗则选取逐层退叠的拱劵,“刻画”窗口的轮楞与线脚。最后,教堂的底层入口区则应用宽大的奥克兰拱劵构成外界形象,并以雾灰实木作为资料形成厚重门板,体积感十足。除却,教堂的外檐材质还采纳红巴黎绿相间的花砖进行砌筑,进而盘活了墙体的肌理档次,并在色彩关系上与天青穹顶形成分明相比。如从仰视的角度去看,教堂传递给人生机勃勃种得体的严穆感与神圣感。

rd_kafka_topic_partition_list_t
  • 所在文件: src/rdkafka.h
  • 用来积攒 rd_kafka_topic_partition_t的可动态扩容的数组
  • 定义:

typedef struct rd_kafka_topic_partition_list_s {
        int cnt;               /**< Current number of elements */ 当前数组中放入的element数量
        int size;              /**< Current allocated size */ // 当前数组的容量
        rd_kafka_topic_partition_t *elems; /**< Element array[] */ 动态数组指针
} rd_kafka_topic_partition_list_t;
  • 扩容操作 rd_kafka_topic_partition_list_grow:

rd_kafka_topic_partition_list_grow (rd_kafka_topic_partition_list_t *rktparlist,
                                    int add_size) {
        if (add_size < rktparlist->size)
                add_size = RD_MAX(rktparlist->size, 32);

        rktparlist->size += add_size;
        // 使用realloc重新分配内存
        rktparlist->elems = rd_realloc(rktparlist->elems,
                                       sizeof(*rktparlist->elems) *
                                       rktparlist->size);

}
  • 创办操作 rd_kafka_topic_partition_list_new:

rd_kafka_topic_partition_list_t *rd_kafka_topic_partition_list_new (int size) {
        rd_kafka_topic_partition_list_t *rktparlist;
        rktparlist = rd_calloc(1, sizeof(*rktparlist));
        rktparlist->size = size;
        rktparlist->cnt = 0;
        if (size > 0)
                rd_kafka_topic_partition_list_grow(rktparlist, size);
        return rktparlist;
}
  • 搜索操作 rd_kafka_topic_partition_list_find:
    topic和partition都等于才好不轻松万分

rd_kafka_topic_partition_list_find (rd_kafka_topic_partition_list_t *rktparlist,
                     const char *topic, int32_t partition) {
    int i = rd_kafka_topic_partition_list_find0(rktparlist,
                            topic, partition);
    if (i == -1)
        return NULL;
    else
        return &rktparlist->elems[i];
}
  • 按索引删除 rd_kafka_topic_partition_list_del_by_idx

rd_kafka_topic_partition_list_del_by_idx (rd_kafka_topic_partition_list_t *rktparlist,
                      int idx) {
    if (unlikely(idx < 0 || idx >= rktparlist->cnt))
        return 0;

        // element数量减1
    rktparlist->cnt--;
        // destory 删除的元素 
    rd_kafka_topic_partition_destroy0(&rktparlist->elems[idx], 0);

        // 作内存的移动, 但不回收
    memmove(&rktparlist->elems[idx], &rktparlist->elems[idx+1],
        (rktparlist->cnt - idx) * sizeof(rktparlist->elems[idx]));

    return 1;
}
  • 排序rd_kafka_topic_partition_list_sort_by_topic
    topic名字不相同按topic名字排,topic名字如出大器晚成辙按partition排

void rd_kafka_topic_partition_list_sort_by_topic (
        rd_kafka_topic_partition_list_t *rktparlist) {
        rd_kafka_topic_partition_list_sort(rktparlist,
                                           rd_kafka_topic_partition_cmp, NULL);
}

下一场呢,在海上捕鱼,日常捕到吗就吃啥,把粥煮到快熟的时候,把捕到的鱼虾蟹等等往锅里精力充沛扔,就成了砂!锅!粥!

     
 在侦察建筑的外界形象之后,笔者起来步入内殿,希图接受“从上而来”的“设计智慧”。推开侧门、轻声而入,高敞的穹顶映重点帘、隐秘的声乐神清气爽,一股敬畏之心情不自禁。怀着此种激情,笔者观望周边,开掘西开教堂的建造平面呈卓越“拉丁十字型”,房内横厅与中厅相冠构成十字廊道。个中心大厅运用“两排单列式方柱”将前殿分成七个部分,即两道侧廊与一齐中廊。而纵深的线性走道又将大家的视觉中央引向圣坛,具有极强的向心力。另外,中厅的柱子共有14根,每排7根,它们通过十字拱劵的连接构成一个完全。而拱劵与方柱还存在龛槽,并援引爱奥尼柱头作为装修。与此相同的时间,中厅两端墙壁还挂有圣经主题素材的油画,甚至使徒宣传教育的日照石雕像,每一种细节的描摹都极度优秀。除却,教堂内部的空中利用鲜青为主基调,以表圣洁之意,他在彩色玻璃的渲染下,光鲜无比。每当周天清早的钟声响起,这里都会坐满真诚的善男信女,共同憧憬着她们心灵中的“天堂”。伴随着这一道道焦点光,作者也渐渐走至圣坛。此时,正逢有一个人长老侍奉祭坛,作者不由向前询问有关教堂的历史,从她的口述中得到消息,西开教堂坐落于原法租界区,由法兰西共和国传教士杜Paul所建,并于一九一八年建设成。因其地处老西开区,毗邻海光寺,故而得名西开教堂。接着他又说:“原先教堂相近存有一大片教会从属建筑,现存的妇产科医院(原天主教医院)、第21中学学(原若瑟会修女法汉高校)、西开小学皆为当下开创,西开教会则为当下的带领与医治工作作出了更加的多进献”。听至此处,小编想在非凡时代,教会既有宣传教育权利,又有社会权利,确实具备发展意义。而他在世纪的前行中,也被津门全员所承受。方今,每年一次的圣诞节此地都汇聚焦多量的人群,此中有善男善女,也可以有普通公众。由此,西开教堂的魅力已不局限于教派层面,而是越发广阔的社会共鸣与观念承认。想到那,小编从耳堂出来,再一次仰望那博大的修造,感慨极其。

rd_kafka_toppar_s
  • 所在文件: src/rdkafka_partition.h
  • 净重数据结构,topic, partition, leader, 生产, 花费,
    各样按期timer都在中间
  • 概念, 那么些布局体巨宏大

struct rd_kafka_toppar_s { /* rd_kafka_toppar_t */
    TAILQ_ENTRY(rd_kafka_toppar_s) rktp_rklink;  /* rd_kafka_t link */
    TAILQ_ENTRY(rd_kafka_toppar_s) rktp_rkblink; /* rd_kafka_broker_t link*/
        CIRCLEQ_ENTRY(rd_kafka_toppar_s) rktp_fetchlink; /* rkb_fetch_toppars */
    TAILQ_ENTRY(rd_kafka_toppar_s) rktp_rktlink; /* rd_kafka_itopic_t link*/
        TAILQ_ENTRY(rd_kafka_toppar_s) rktp_cgrplink;/* rd_kafka_cgrp_t link */
        rd_kafka_itopic_t       *rktp_rkt;
        shptr_rd_kafka_itopic_t *rktp_s_rkt;  /* shared pointer for rktp_rkt */
    int32_t            rktp_partition;
        //LOCK: toppar_lock() + topic_wrlock()
        //LOCK: .. in partition_available()
        int32_t            rktp_leader_id;   /**< Current leader broker id.
                                              *   This is updated directly
                                              *   from metadata. */
    rd_kafka_broker_t *rktp_leader;      /**< Current leader broker
                                              *   This updated asynchronously
                                              *   by issuing JOIN op to
                                              *   broker thread, so be careful
                                              *   in using this since it
                                              *   may lag. */
        rd_kafka_broker_t *rktp_next_leader; /**< Next leader broker after
                                              *   async migration op. */
    rd_refcnt_t        rktp_refcnt;
    mtx_t              rktp_lock;
 rd_atomic32_t      rktp_version;         /* Latest op version.
                                                  * Authoritative (app thread)*/
    int32_t            rktp_op_version;      /* Op version of curr command
                          * state from.
                          * (broker thread) */
        int32_t            rktp_fetch_version;   /* Op version of curr fetch.
                                                    (broker thread) */

    enum {
        RD_KAFKA_TOPPAR_FETCH_NONE = 0,
                RD_KAFKA_TOPPAR_FETCH_STOPPING,
                RD_KAFKA_TOPPAR_FETCH_STOPPED,
        RD_KAFKA_TOPPAR_FETCH_OFFSET_QUERY,
        RD_KAFKA_TOPPAR_FETCH_OFFSET_WAIT,
        RD_KAFKA_TOPPAR_FETCH_ACTIVE,
    } rktp_fetch_state;    
int32_t            rktp_fetch_msg_max_bytes; /* Max number of bytes to
                                                      * fetch.
                                                      * Locality: broker thread
                                                      */

        rd_ts_t            rktp_ts_fetch_backoff; /* Back off fetcher for
                                                   * this partition until this
                                                   * absolute timestamp
                                                   * expires. */

    int64_t            rktp_query_offset;    /* Offset to query broker for*/
    int64_t            rktp_next_offset;     /* Next offset to start
                                                  * fetching from.
                                                  * Locality: toppar thread */
    int64_t            rktp_last_next_offset; /* Last next_offset handled
                           * by fetch_decide().
                           * Locality: broker thread */
    int64_t            rktp_app_offset;      /* Last offset delivered to
                          * application + 1 */
    int64_t            rktp_stored_offset;   /* Last stored offset, but
                          * maybe not committed yet. */
        int64_t            rktp_committing_offset; /* Offset currently being
                                                    * committed */
    int64_t            rktp_committed_offset; /* Last committed offset */
    rd_ts_t            rktp_ts_committed_offset; /* Timestamp of last
                                                      * commit */

        struct offset_stats rktp_offsets; /* Current offsets.
                                           * Locality: broker thread*/
        struct offset_stats rktp_offsets_fin; /* Finalized offset for stats.
                                               * Updated periodically
                                               * by broker thread.
                                               * Locks: toppar_lock */

    int64_t rktp_hi_offset;              /* Current high offset.
                          * Locks: toppar_lock */
        int64_t rktp_lo_offset;         
 rd_ts_t            rktp_ts_offset_lag;

    char              *rktp_offset_path;     /* Path to offset file */
    FILE              *rktp_offset_fp;       /* Offset file pointer */
        rd_kafka_cgrp_t   *rktp_cgrp;            /* Belongs to this cgrp */

        int                rktp_assigned;   /* Partition in cgrp assignment */

        rd_kafka_replyq_t  rktp_replyq; /* Current replyq+version
                     * for propagating
                     * major operations, e.g.,
                     * FETCH_STOP. */
    int                rktp_flags;

        shptr_rd_kafka_toppar_t *rktp_s_for_desp; /* Shared pointer for
                                                   * rkt_desp list */
        shptr_rd_kafka_toppar_t *rktp_s_for_cgrp; /* Shared pointer for
                                                   * rkcg_toppars list */
        shptr_rd_kafka_toppar_t *rktp_s_for_rkb;  /* Shared pointer for
                                                   * rkb_toppars list */

    /*
     * Timers
     */
    rd_kafka_timer_t rktp_offset_query_tmr;  /* Offset query timer */
    rd_kafka_timer_t rktp_offset_commit_tmr; /* Offset commit timer */
    rd_kafka_timer_t rktp_offset_sync_tmr;   /* Offset file sync timer */
        rd_kafka_timer_t rktp_consumer_lag_tmr;  /* Consumer lag monitoring
                          * timer */

        int rktp_wait_consumer_lag_resp;         /* Waiting for consumer lag
                                                  * response. */

    struct {
        rd_atomic64_t tx_msgs;
        rd_atomic64_t tx_bytes;
                rd_atomic64_t msgs;
                rd_atomic64_t rx_ver_drops;
    } rktp_c;
}
  • 创立叁个 rd_kafka_toppar_t对象 rd_kafka_toppar_new0:

shptr_rd_kafka_toppar_t *rd_kafka_toppar_new0 (rd_kafka_itopic_t *rkt,
                           int32_t partition,
                           const char *func, int line) {
    rd_kafka_toppar_t *rktp;

       // 分配内存
    rktp = rd_calloc(1, sizeof(*rktp));

        // 各项赋值
    rktp->rktp_partition = partition;

        // 属于哪个topic
    rktp->rktp_rkt = rkt;

        rktp->rktp_leader_id = -1;
    rktp->rktp_fetch_state = RD_KAFKA_TOPPAR_FETCH_NONE;
        rktp->rktp_fetch_msg_max_bytes
            = rkt->rkt_rk->rk_conf.fetch_msg_max_bytes;
    rktp->rktp_offset_fp = NULL;
        rd_kafka_offset_stats_reset(&rktp->rktp_offsets);
        rd_kafka_offset_stats_reset(&rktp->rktp_offsets_fin);
        rktp->rktp_hi_offset = RD_KAFKA_OFFSET_INVALID;
    rktp->rktp_lo_offset = RD_KAFKA_OFFSET_INVALID;
    rktp->rktp_app_offset = RD_KAFKA_OFFSET_INVALID;
        rktp->rktp_stored_offset = RD_KAFKA_OFFSET_INVALID;
        rktp->rktp_committed_offset = RD_KAFKA_OFFSET_INVALID;
    rd_kafka_msgq_init(&rktp->rktp_msgq);
        rktp->rktp_msgq_wakeup_fd = -1;
    rd_kafka_msgq_init(&rktp->rktp_xmit_msgq);
    mtx_init(&rktp->rktp_lock, mtx_plain);

        rd_refcnt_init(&rktp->rktp_refcnt, 0);
    rktp->rktp_fetchq = rd_kafka_q_new(rkt->rkt_rk);
        rktp->rktp_ops    = rd_kafka_q_new(rkt->rkt_rk);
        rktp->rktp_ops->rkq_serve = rd_kafka_toppar_op_serve;
        rktp->rktp_ops->rkq_opaque = rktp;
        rd_atomic32_init(&rktp->rktp_version, 1);
    rktp->rktp_op_version = rd_atomic32_get(&rktp->rktp_version);

        // 开始一个timer, 来定时统计消息的lag情况, 目前看是一个`rd_kafka_toppar_t`对象就一个timer, 太多了, 可以用时间轮来作所有partiton的timer
        if (rktp->rktp_rkt->rkt_rk->rk_conf.stats_interval_ms > 0 &&
            rkt->rkt_rk->rk_type == RD_KAFKA_CONSUMER &&
            rktp->rktp_partition != RD_KAFKA_PARTITION_UA) {
                int intvl = rkt->rkt_rk->rk_conf.stats_interval_ms;
                if (intvl < 10 * 1000 /* 10s */)
                        intvl = 10 * 1000;
        rd_kafka_timer_start(&rkt->rkt_rk->rk_timers,
                     &rktp->rktp_consumer_lag_tmr,
                                     intvl * 1000ll,
                     rd_kafka_toppar_consumer_lag_tmr_cb,
                     rktp);
        }

        rktp->rktp_s_rkt = rd_kafka_topic_keep(rkt);

        // 设置其fwd op queue到rd_kakfa_t中的rd_ops, 这样这个rd_kafka_toppar_t对象用到的ops_queue就是rd_kafka_t的了
    rd_kafka_q_fwd_set(rktp->rktp_ops, rkt->rkt_rk->rk_ops);
    rd_kafka_dbg(rkt->rkt_rk, TOPIC, "TOPPARNEW", "NEW %s [%"PRId32"] %p (at %s:%d)",
             rkt->rkt_topic->str, rktp->rktp_partition, rktp,
             func, line);

    return rd_kafka_toppar_keep_src(func, line, rktp);
}
  • 销毁三个rd_kafka_toppar_t对象rd_kafka_toppar_destroy_final

void rd_kafka_toppar_destroy_final (rd_kafka_toppar_t *rktp) {
        // 停掉相应的timer, 清空ops queue
        rd_kafka_toppar_remove(rktp);

        // 将msgq中的kafka message回调给app层后清空
    rd_kafka_dr_msgq(rktp->rktp_rkt, &rktp->rktp_msgq,
             RD_KAFKA_RESP_ERR__DESTROY);
    rd_kafka_q_destroy_owner(rktp->rktp_fetchq);
        rd_kafka_q_destroy_owner(rktp->rktp_ops);

    rd_kafka_replyq_destroy(&rktp->rktp_replyq);

    rd_kafka_topic_destroy0(rktp->rktp_s_rkt);

    mtx_destroy(&rktp->rktp_lock);

        rd_refcnt_destroy(&rktp->rktp_refcnt);

    rd_free(rktp);
}
  • 从一个rd_kafka_itopic_t(那一个大家前面会有极度篇章来介绍,
    这里只要求掌握它代表topic就可以,
    里面包蕴属于它的parition列表)获取钦点parition:

shptr_rd_kafka_toppar_t *rd_kafka_toppar_get0 (const char *func, int line,
                                               const rd_kafka_itopic_t *rkt,
                                               int32_t partition,
                                               int ua_on_miss) {
        shptr_rd_kafka_toppar_t *s_rktp;

        // 数组索引下标来获取 partition
    if (partition >= 0 && partition < rkt->rkt_partition_cnt)
        s_rktp = rkt->rkt_p[partition];
    else if (partition == RD_KAFKA_PARTITION_UA || ua_on_miss)
        s_rktp = rkt->rkt_ua;
    else
        return NULL;

    if (s_rktp)
               // 引用计数加1 
                return rd_kafka_toppar_keep_src(func,line,
                                                rd_kafka_toppar_s2i(s_rktp));

    return NULL;
}
  • 按topic名字和partition来得到叁个rd_kafka_toppar_t对象,
    未有找到topic, 就先创建那个 rd_kafka_itopic_t对象

shptr_rd_kafka_toppar_t *rd_kafka_toppar_get2 (rd_kafka_t *rk,
                                               const char *topic,
                                               int32_t partition,
                                               int ua_on_miss,
                                               int create_on_miss) {
    shptr_rd_kafka_itopic_t *s_rkt;
        rd_kafka_itopic_t *rkt;
        shptr_rd_kafka_toppar_t *s_rktp;

        rd_kafka_wrlock(rk);

        /* Find or create topic */
        // 所有的 rd_kafka_itopic_t对象都存在rd_kafka_t的rkt_topic的tailq队列里, 这里先查找
    if (unlikely(!(s_rkt = rd_kafka_topic_find(rk, topic, 0/*no-lock*/)))) {
                if (!create_on_miss) {
                        rd_kafka_wrunlock(rk);
                        return NULL;
                }
                // 没找到就先创建  rd_kafka_itopic_t对象
                s_rkt = rd_kafka_topic_new0(rk, topic, NULL,
                        NULL, 0/*no-lock*/);
                if (!s_rkt) {
                        rd_kafka_wrunlock(rk);
                        rd_kafka_log(rk, LOG_ERR, "TOPIC",
                                     "Failed to create local topic \"%s\": %s",
                                     topic, rd_strerror(errno));
                        return NULL;
                }
        }

        rd_kafka_wrunlock(rk);

        rkt = rd_kafka_topic_s2i(s_rkt);

    rd_kafka_topic_wrlock(rkt);
    s_rktp = rd_kafka_toppar_desired_add(rkt, partition);
    rd_kafka_topic_wrunlock(rkt);

        rd_kafka_topic_destroy0(s_rkt);

    return s_rktp;
}
  • desired partition: desired partition状态的parititon,
    源码中的解释如下:

The desired partition list is the list of partitions that are
desired
(e.g., by the consumer) but not yet seen on a broker.
As soon as the partition is seen on a broker the toppar is moved
from
the desired list and onto the normal rkt_p array.
When the partition on the broker goes away a desired partition is
put
back on the desired list

粗略说正是急需某贰个partition,
可是其豆蔻梢头parition的具体新闻还没从broker拿掉,那样的parition正是desired
parition, 在rd_kafka_itopic_t中有四个rkt_desp的list,
特地用来存这么的parition, 针对其有如下几个操作,都比较轻松:

rd_kafka_toppar_desired_get
rd_kafka_toppar_desired_link
rd_kafka_toppar_desired_unlink
rd_kafka_toppar_desired_add0
rd_kafka_toppar_desired_add
rd_kafka_toppar_desired_del
  • partition在broker间迁移rd_kafka_toppar_broker_migrate:

static void rd_kafka_toppar_broker_migrate (rd_kafka_toppar_t *rktp,
                                            rd_kafka_broker_t *old_rkb,
                                            rd_kafka_broker_t *new_rkb) {
        rd_kafka_op_t *rko;
        rd_kafka_broker_t *dest_rkb;
        int had_next_leader = rktp->rktp_next_leader ? 1 : 0;

        /* Update next leader */
        if (new_rkb)
                rd_kafka_broker_keep(new_rkb);
        if (rktp->rktp_next_leader)
                rd_kafka_broker_destroy(rktp->rktp_next_leader);
        rktp->rktp_next_leader = new_rkb;

        // 在迁移没完成时有可能再次迁移了, 这个时候是不是需要加锁? 
        if (had_next_leader)
                return;

    if (rktp->rktp_fetch_state == RD_KAFKA_TOPPAR_FETCH_OFFSET_WAIT) {
        rd_kafka_toppar_set_fetch_state(
            rktp, RD_KAFKA_TOPPAR_FETCH_OFFSET_QUERY);
        rd_kafka_timer_start(&rktp->rktp_rkt->rkt_rk->rk_timers,
                     &rktp->rktp_offset_query_tmr,
                     500*1000,
                     rd_kafka_offset_query_tmr_cb,
                     rktp);
    }

        //  迁移前broker放到LEAVE op
        if (old_rkb) {
                rko = rd_kafka_op_new(RD_KAFKA_OP_PARTITION_LEAVE);
                dest_rkb = old_rkb;
        } else {
                /* No existing broker, send join op directly to new leader. */
                rko = rd_kafka_op_new(RD_KAFKA_OP_PARTITION_JOIN);
                dest_rkb = new_rkb;
        }

        rko->rko_rktp = rd_kafka_toppar_keep(rktp);

        rd_kafka_q_enq(dest_rkb->rkb_ops, rko);
}
  • broker的delegate操作:

void rd_kafka_toppar_broker_delegate (rd_kafka_toppar_t *rktp,
                      rd_kafka_broker_t *rkb,
                      int for_removal) {
        rd_kafka_t *rk = rktp->rktp_rkt->rkt_rk;
        int internal_fallback = 0;

        /* Delegate toppars with no leader to the
         * internal broker for bookkeeping. */
        // 如果迁移到的broker是NULL, 就获取一个internal broker -> rkb
        if (!rkb && !for_removal && !rd_kafka_terminating(rk)) {
                rkb = rd_kafka_broker_internal(rk);
                internal_fallback = 1;
        }

    if (rktp->rktp_leader == rkb && !rktp->rktp_next_leader) {
                rd_kafka_dbg(rktp->rktp_rkt->rkt_rk, TOPIC, "BRKDELGT",
                 "%.*s [%"PRId32"]: not updating broker: "
                             "already on correct broker %s",
                 RD_KAFKAP_STR_PR(rktp->rktp_rkt->rkt_topic),
                 rktp->rktp_partition,
                             rkb ? rd_kafka_broker_name(rkb) : "(none)");

                if (internal_fallback)
                        rd_kafka_broker_destroy(rkb);
        return;
        }

        // 实际的迁移操作
        if (rktp->rktp_leader || rkb)
                rd_kafka_toppar_broker_migrate(rktp, rktp->rktp_leader, rkb);

        if (internal_fallback)
                rd_kafka_broker_destroy(rkb);
}
  • 提交offstet到broker rd_kafka_toppar_offset_commit

void rd_kafka_toppar_offset_commit (rd_kafka_toppar_t *rktp, int64_t offset,
                    const char *metadata) {
        rd_kafka_topic_partition_list_t *offsets;
        rd_kafka_topic_partition_t *rktpar;

        // 构造 一个rd_kafka_topic_partition_list, 把当前的topic添加进去, 包括要提交的offset
        offsets = rd_kafka_topic_partition_list_new(1);
        rktpar = rd_kafka_topic_partition_list_add(
                offsets, rktp->rktp_rkt->rkt_topic->str, rktp->rktp_partition);
        rktpar->offset = offset;
        if (metadata) {
                rktpar->metadata = rd_strdup(metadata);
                rktpar->metadata_size = strlen(metadata);
        }

        // rd_kafka_toppar_t对象更新rktp_committing_offset,表示正在提交的offset
        rktp->rktp_committing_offset = offset;

       // 异步提交offset, 这个操作在之后介绍kafka consumer是会详细分析
        rd_kafka_commit(rktp->rktp_rkt->rkt_rk, offsets, 1/*async*/);

        rd_kafka_topic_partition_list_destroy(offsets);
}
  • 安装下二回拉取数据时开始的offset地方,即rd_kafka_toppar_trktp_next_offset

void rd_kafka_toppar_next_offset_handle (rd_kafka_toppar_t *rktp,
                                         int64_t Offset) {
        // 如果Offset是BEGINNING,END, 发起一个rd_kafka_toppar_offset_request操作,从broker获取offset
        // 如果Offset是RD_KAFKA_OFFSET_INVALID, 需要enqueue一个error op, 设置fetch状态为RD_KAFKA_TOPPAR_FETCH_NONE
        if (RD_KAFKA_OFFSET_IS_LOGICAL(Offset)) {
                /* Offset storage returned logical offset (e.g. "end"),
                 * look it up. */
                rd_kafka_offset_reset(rktp, Offset, RD_KAFKA_RESP_ERR_NO_ERROR,
                                      "update");
                return;
        }

        /* Adjust by TAIL count if, if wanted */
        // 获取从tail开始往前推cnt个offset的位置
        if (rktp->rktp_query_offset <=
            RD_KAFKA_OFFSET_TAIL_BASE) {
                int64_t orig_Offset = Offset;
                int64_t tail_cnt =
                        llabs(rktp->rktp_query_offset -
                              RD_KAFKA_OFFSET_TAIL_BASE);

                if (tail_cnt > Offset)
                        Offset = 0;
                else
                        Offset -= tail_cnt;
        }

        //设置rktp_next_offset
        rktp->rktp_next_offset = Offset;

        rd_kafka_toppar_set_fetch_state(rktp, RD_KAFKA_TOPPAR_FETCH_ACTIVE);

        /* Wake-up broker thread which might be idling on IO */
        if (rktp->rktp_leader)
                rd_kafka_broker_wakeup(rktp->rktp_leader);

}
  • 从coordinattor获取已交给的offset(FetchOffsetRequest)
    rd_kafka_toppar_offset_fetch:

void rd_kafka_toppar_offset_fetch (rd_kafka_toppar_t *rktp,
                                   rd_kafka_replyq_t replyq) {
        rd_kafka_t *rk = rktp->rktp_rkt->rkt_rk;
        rd_kafka_topic_partition_list_t *part;
        rd_kafka_op_t *rko;

        part = rd_kafka_topic_partition_list_new(1);
        rd_kafka_topic_partition_list_add0(part,
                                           rktp->rktp_rkt->rkt_topic->str,
                                           rktp->rktp_partition,
                       rd_kafka_toppar_keep(rktp));

        // 构造OffsetFetch的operator
        rko = rd_kafka_op_new(RD_KAFKA_OP_OFFSET_FETCH);
    rko->rko_rktp = rd_kafka_toppar_keep(rktp);
    rko->rko_replyq = replyq;

    rko->rko_u.offset_fetch.partitions = part;
    rko->rko_u.offset_fetch.do_free = 1;

        // OffsetFetch 请求是与消费有关的,放入cgrp的op queue里
        rd_kafka_q_enq(rktp->rktp_cgrp->rkcg_ops, rko);
}
  • 获得用于花费的可行的offset

void rd_kafka_toppar_offset_request (rd_kafka_toppar_t *rktp,
                     int64_t query_offset, int backoff_ms) {
    rd_kafka_broker_t *rkb;
        rkb = rktp->rktp_leader;

         // 如果rkb是无效的,需要下一个timer来定时query
        if (!backoff_ms && (!rkb || rkb->rkb_source == RD_KAFKA_INTERNAL))
                backoff_ms = 500;

        if (backoff_ms) {
                rd_kafka_toppar_set_fetch_state(
                        rktp, RD_KAFKA_TOPPAR_FETCH_OFFSET_QUERY);
                // 启动timer, timer到期会执行rd_kafka_offset_query_tmr_cb回调,这个回调还是调用当前这个函数
        rd_kafka_timer_start(&rktp->rktp_rkt->rkt_rk->rk_timers,
                     &rktp->rktp_offset_query_tmr,
                     backoff_ms*1000ll,
                     rd_kafka_offset_query_tmr_cb, rktp);
        return;
        }

        // stop这个重试的timer
        rd_kafka_timer_stop(&rktp->rktp_rkt->rkt_rk->rk_timers,
                            &rktp->rktp_offset_query_tmr, 1/*lock*/);

        // 从coordinattor获取需要消费的offset
    if (query_offset == RD_KAFKA_OFFSET_STORED &&
            rktp->rktp_rkt->rkt_conf.offset_store_method ==
            RD_KAFKA_OFFSET_METHOD_BROKER) {
                /*
                 * Get stored offset from broker based storage:
                 * ask cgrp manager for offsets
                 */
                rd_kafka_toppar_offset_fetch(
            rktp,
            RD_KAFKA_REPLYQ(rktp->rktp_ops,
                    rktp->rktp_op_version));

    } else {
                shptr_rd_kafka_toppar_t *s_rktp;
                rd_kafka_topic_partition_list_t *offsets;

                /*
                 * Look up logical offset (end,beginning,tail,..)
                 */
                s_rktp = rd_kafka_toppar_keep(rktp);

        if (query_offset <= RD_KAFKA_OFFSET_TAIL_BASE)
            query_offset = RD_KAFKA_OFFSET_END;

                offsets = rd_kafka_topic_partition_list_new(1);
                rd_kafka_topic_partition_list_add(
                        offsets,
                        rktp->rktp_rkt->rkt_topic->str,
                        rktp->rktp_partition)->offset = query_offset;

                // 基本上用于reset offset, 获取当前partition的最旧offset或最新offset
                rd_kafka_OffsetRequest(rkb, offsets, 0,
                                       RD_KAFKA_REPLYQ(rktp->rktp_ops,
                                                       rktp->rktp_op_version),
                                       rd_kafka_toppar_handle_Offset,
                                       s_rktp);

                rd_kafka_topic_partition_list_destroy(offsets);
        }

        rd_kafka_toppar_set_fetch_state(rktp,
                RD_KAFKA_TOPPAR_FETCH_OFFSET_WAIT);
}

好了,做法大约也就那样,今土耳其共和国(Türkiye Cumhuriyeti)语章就写到那了。感激我们。

     
 西开教堂他是自笔者陶醉夜下的“锡安城”;是霓虹灯下的“静默标杆”,是经济贸易大街的“绿洲岛”,又是安慰心灵的“祷告室”。不管您信与不相信,爱与不爱,他平昔“伫立”在那,“守候”在那边,又在这里边“等待”,用她温柔的膀子招待“流浪的游子”“回家”。笔者想教堂的得体与盛大只是情势的结构,其内在的感召力才是功能的实业,创设出大器晚成种心灵上的“宫室”才是教堂设计的初志。

Librdkafka源码深入分析-Content Table

实质上小编想说的是,砂锅粥里面下什么料,真的能够十分轻巧。

但既然要讲咋做,大家就必得来掰黄金时代掰黄金年代款最特出的,虾蟹砂锅粥。其实出入无间,做法都大约。

明天买了四只小膏蟹,半斤九节虾,大约4-5人份砂锅粥。大概把资料说下:

膏蟹 2~3只

虾 约半斤

珍珠米 2米杯

干贝 若干

香菇 4~5个

姜丝 若干

芹菜、香菜、冬菜 若干

虾蟹先洗洗。

做这么些前边,先泡几个花菇与干贝,但要分开泡。然后2杯米洗净后泡水,泡起码半钟头以上。籼米是比较符合的米。

泡米的还要,大家来管理虾蟹。

都是活的,咋整呢。先来搞稻蟹吧。

怎么着把胜芳蟹搞死,猜想那难题要愁死了一批小白,其实很简短,把方蟹翻过来,在石蟹后盖尖处,找把剪刀或竹筷之类的利器,插进去,默念10分钟,大概就死大致了。

何以吗?因为此处是蟹心。

哪些判断蟹是或不是早就死了啊?当蟹腿已经铺开,碰它也不动,就印证已经死了。

那儿能够给它包扎,然后初始分尸。

笔者们先把后盖给剪下来,后盖连接处为蟹肠,记得漱口一下。不洗的话,或者会有淡淡的女华香。

剪掉全部脚尖。不剪也行,只不过没肉,本身又嫌脏,干脆剪掉。

掀开蟹壳,去蟹鳃、蟹心与蟹胃。此中蟹心极寒,应当要去掉。

蟹鳃相当轻松找,图中所示多个地方的事物直接摘除干净就可以。

蟹胃则要求找一下,最简易方法是用剪刀或镊子,插入图中所示蟹胃地点,夹紧百尺竿头扯,扯出来的那块东西正是蟹胃。

蟹心则是在蟹鳃中间的一小块,抠掉就能够。

蟹切两半,分尸后大约成那样。

石蟹作为腐食动物,属于比较穷的这种,通常吃土啥的,建议拿个放弃牙刷,把石蟹刷精神激昂刷,依旧综上可得能够认为到刷掉非常多污秽的。可是小心别把蟹膏给冲走了。

有被蟹夹过心境阴影面积一点都不小的同桌,避防万蒸蒸日上,也足以在分尸前,先把多少个大钳剪下来。

剪的时候手不要抓着钳子,就算死了,出于反射,剪的时候钳子依旧会夹一下的,被夹到仍旧有一些痛的,别问作者怎么精晓。

接下去管理虾,把虾头上的枪剪了,然后把虾开边,去虾的肠道。

把虾蟹撒一点点盐稍微腌渍一下。虾蟹尽管大约管理完成。

管理完虾蟹大家承接切配料。

蟹是相比寒凉的,供给用姜丝去寒,所以也要切点姜丝。

下一场把泡好的香信切成条,切点水芹粒、胡荽碎、冬菜碎备用。

接下去起头煮粥。砂锅相符煮粥,因为受热均匀,不易糊底,家里有砂锅最佳,未有的话,选个铸铁材料的也得以,因为铸铁也是受热相比均匀的质量。

水须求一遍加足,文火煮沸,加入泡好的米及少些大豆油,葡萄籽油能够削减粥水泡沫溢出。煮粥时不经常供给搅拌一下,防止糊底。

待米粒开花,转中火,加入元贝、蟹、姜丝、花菇,煮约5分钟。虾轻便过熟,所以先不下。

再转温火,插手虾、冬菜碎,再煮5分钟关火。

那会儿应当粥水相比浓稠了,最后参预延荽碎和水芹粒,加点盐调味,那砂锅粥固然达成了。

盛一碗,雪人蟹君还柔情脉脉地瞅着您,是或不是马上感觉内心暖暖的?

有关调味,某人会再用玉椒、鱼露、老抽等如日方升种或三种调味剂,其实这些就看个人喜好了,只倘若您赏识的含意正是正宗的,不必拘泥太多。笔者要好喜好只用盐,单纯且能够突出原味鲜甜。

有关汤底,也足以先用脊椎骨煮成汤,再拿冬瓜汤煮粥,那样粥会越来越香。然而明日自己虾蟹买多了,就不下脊椎骨了。

其他煮的进度中,或许会并发粥太稠的难点。假使中途须求加水,必需加热水,切记不可加冷水。

= 步骤 =

泡花菇与干贝,米泡三十分钟以上。

管理虾蟹,并略微腌渍

切香菌、姜丝、西芹粒、香菜碎与冬菜碎。

白热水加入一点点山茶油,大火煮粥至米粒开花。

转中火,加入元贝、蟹、姜丝、冬菇,煮约5分钟。

转大火,到场虾、冬菜碎,再煮5分钟后关火。

投入胡荽碎与香芹粒,加盐调味。

相关文章