又被问到Kafka啦？一起来整理思维导图吧！

阳光穿透心脏的1/2处 2023-02-13 11:14 11阅读 0赞

最近在对照 《Kafka 权威指南》 这本书, 整理kafka 的一些细枝末叶。随着文章发布，也会逐步完善一份思维导图。没有耐心的小伙伴可以直接跳到文章的最后看总结， 或者可以关注我新开的公众号：程序员BenJi， 回复 ：kafka即可获取。

## 1 创建Kafka生产者 ##

首先要创建一个生产者对象, 并设置一些属性. Kafka 生产者有**3个必选的属性**, \*\*一定要记得， 不然面试官会觉得你很水！\*\*分别为

# broker的地址清单,清单里不需要提供所有broker的地址, 
     # 生产者会从给定的broker 中查找 其他的broker信息.
    bootstraps.servers
    key.serializer
    value.serializer

### 1.2 其他可选属性 ###

关于这些属性， **要知道不同属性会对系统有什么影响**  
比如 ack (0,1,-1) , retries ,超时时间 等, 详见 思维导图的 备注说明  
这边因为篇幅有限, 讲解一下常常面试被问到的 ack属性

#### 1.2.1 ack属性（设置broker收到消息的应答行为）： ####

*  ack == 0  
    生产者成功写入消息之前,不会等待任何来自服务器的消息,**可能会造成消息丢失**
 *  ack == 1  
    只等待集群首领节点接收到消息便返回.  
    如果首领节点尚未同步数据到其他集群节点, **容易造成数据丢失**
 *  ack == -1  
    需要集群中所有副本同步消息. 会**影响到发送消息的效率**

#### 1.2.2 Retries属性（设置生产者发送消息后，异常重试次数）： ####

生产者发送消息出现异常后, 尝试重试的次数。**当这个值设置大于1的时候，可能会引起顺序消费的问题。**  
我们都知道kafka 保证消息在同一个分区内是有序的,但是如果第一条消息发生重试, 第二条消息写入成功。随后第一条消息重试成功, 这样**消息顺序就会发生错误**。  
要想知道如何解决由重试引发的消息顺序性问题，发送Kafka到公众号，获取思维导图源码，查看max.in.flight.request.per.connection属性设置吧， 哈哈！

#### 1.2.2 其他属性 ####

这边因为篇幅有限, retries ,超时时间 等, 详见文末的 思维导图的 备注说明

## 2 发送消息(同步,异步) ##

/**
         * 发送演示
         * 发送消息必传参数:Topic, Value , 可选参数 partition , key
         */
        // 最简单的直接发送
        private void sendMessage(){
            // 构造消息, 必穿参数Topic , Value
            ProducerRecord<String, String> record = new ProducerRecord<>("test-topic", 0, "key", "v");
            try {
                RecordMetadata recordMetadata = producer.send(record).get();
                // 返回的元数据里面包含重要的 offset , partition 数据
                recordMetadata.offset();
                recordMetadata.partition();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    
        // 异步发送, 需要提供回调
        private void sendMessageAsyn(){
            ProducerRecord<String, String> record = new ProducerRecord<>("test-topic", 0, "key", "v");
            producer.send(record,new CallBackDemo());
        }
    
        // 实现kafak 的 CallBack 接口
        class CallBackDemo implements Callback{
            @Override
            public void onCompletion(RecordMetadata metadata, Exception exception) {
                if(exception != null){
                    exception.printStackTrace();
                }
            }
        }

### 2.1 自定义分区器 ###

kafka 的分区, 具体指的是消息分配到哪一个partition.

1.  如果采用默认的分区器.

*  当不设定key, 或者key设定为null, 消息会通过轮询机制, **平均分配到每个partition**.
 *  当指定了key, 那么分区器会**通过对key的散列计算出来的值, 分配到指定的某个partition**.

1.  实际使用中, 我们往往不希望某一个key 的数据依据默认的散列值算法. 和其他消息分配到同一个分区.这个时候就可以对key采用**自定义分区**

/**
         * 自定义 Taobao分区器 
         * 分区器需要实现 Kafka 的 Partitioner 接口
         * 重写 partion 方法
         */
        public class  BananaPartitioner implements Partitioner{
            // 演示了 key == Taobao 的消息, 总是分配到最后一个分区,
            // 其他消息 散列到除了最后一个分区之外的其他分区
            // 需要返回分区 idx
            @Override
            public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
                List<PartitionInfo> partitionInfos = cluster.partitionsForTopic(topic);
                int lastPartitionNumber = partitionInfos.size();
                // 如果key == banner ,  总是分配到最后一个分区,
                if ("Taobao".equalsIgnoreCase((String) key)){
                    return lastPartitionNumber;
                }
                // 其他消息 散列到除了最后一个分区之外的其他分区
                return (Math.abs(Utils.murmur2(keyBytes))%(lastPartitionNumber - 1));
            }
    
            @Override
            public void close() {
    
            }
    
            @Override
            public void configure(Map<String, ?> configs) {
    
            }
        }

--------------------

## 总结一下， kafka 的生产者Api： ##

1.  创建生产者，设置不同属性，必传参数：bootstraps.servers,key.serializer,value.serializer；
2.  send（） 方法发送消息，分为同步异步。 必传参数Topic, value ,可选参数 partition 和 key：
3.  自定义分区器，描述指定Key 发送到哪个Partition上。

下面是我整理的思维导图，源文件请关注我新建的公众号： **程序员BenJi ， 回复 kafka 即可获取：**  
定期分享优质的知识思维导图，帮助大家建立结构化的知识体系，欢迎关注~  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzE0ODUzODg5_size_16_color_FFFFFF_t_70]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzE0ODUzODg5_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/2020052821445798.jpg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzE0ODUzODg5,size_16,color_FFFFFF,t_70