springboot第69集：字节跳动后端二面经，一文让你走出微服务迷雾架构周刊

1.  简介

1.1  消息队列简介

1.1.1  什么是消息队列

消息队列，英文名：Message Queue，经常缩写为MQ。从字面上来理解，消息队列是一种用来存储消息的队列。来看一下下面的代码：

// 1. 创建一个保存字符串的队列
QueuestringQueue = new LinkedList();

// 2. 往消息队列中放入消息
stringQueue.offer( "hello" );

// 3. 从消息队列中取出消息并打印
System.out.println(stringQueue.poll());

上述代码，创建了一个队列，先往队列中添加了一个消息，然后又从队列中取出了一个消息。这说明了队列是可以用来存取消息的。

我们可以简单理解消息队列就是将需要传输的数据存放在队列中。

1.1.1  消息队列中间件

消息队列中间件就是用来存储消息的软件（组件）。举个例子来理解，为了分析网站的用户行为，我们需要记录用户的访问日志。这些一条条的日志，可以看成是一条条的消息，我们可以将它们保存到消息队列中。将来有一些应用程序需要处理这些日志，就可以随时将这些消息取出来处理。

目前市面上的消息队列有很多，例如：Kafka、RabbitMQ、ActiveMQ、RocketMQ、ZeroMQ等。

消息队列的应用场景

电商网站中，新的用户注册时，需要将用户的信息保存到数据库中，同时还需要额外发送注册的邮件通知、以及短信注册码给用户。但因为发送邮件、发送注册短信需要连接外部的服务器，需要额外等待一段时间，此时，就可以使用消息队列来进行异步处理，从而实现快速响应。

1.1.1.1  日志处理（大数据领域常见）

大型电商网站（淘宝、京东、国美、苏宁...）、App（抖音、美团、滴滴等）等需要分析用户行为，要根据用户的访问行为来发现用户的喜好以及活跃情况，需要在页面上收集大量的用户访问信息。

点对点模式特点：

每个消息只有一个接收者（Consumer）(即一旦被消费，消息就不再在消息队列中)

发送者和接收者间没有依赖性，发送者发送消息之后，不管有没有接收者在运行，都不会影响到发送者下次发送消息；

接收者在成功接收消息之后需向队列应答成功，以便消息队列删除当前接收的消息；

Kafka是由Apache软件基金会开发的一个开源流平台，由Scala和Java编写。Kafka的Apache官网是这样介绍Kakfa的。

Apache Kafka是一个分布式流平台。一个分布式的流平台应该包含3点关键的能力：

1. 发布和订阅流数据流，类似于消息队列或者是企业消息传递系统

2. 以容错的持久化方式存储数据流

处理数据流

1. Publish and subscribe：发布与订阅

2. Store：存储

3. Process：处理

我们通常将Apache Kafka用在两类程序：

1. 建立实时数据管道，以可靠地在系统或应用程序之间获取数据

2. 构建实时流应用程序，以转换或响应数据流

上图，我们可以看到：

1. Producers：可以有很多的应用程序，将消息数据放入到Kafka集群中。

2. Consumers：可以有很多的应用程序，将消息数据从Kafka集群中拉取出来。

3. Connectors：Kafka的连接器可以将数据库中的数据导入到Kafka，也可以将Kafka的数据导出到

数据库中。

4. Stream Processors：流处理器可以Kafka中拉取数据，也可以将数据写入到Kafka中。

Kafka比ActiveMQ牛逼得多

可以注意到Kafka的版本号为：kafka_2.12-2.4.1，因为kafka主要是使用scala语言开发的，2.12为scala的版本号。http://kafka.apache.org/downloads可以查看到每个版本的发布时间。

基于1个分区1个副本的基准测试

测试步骤：

1. 启动Kafka集群

2. 创建一个1个分区1个副本的topic: benchmark

3. 同时运行生产者、消费者基准测试程序

4. 观察结果

<**repositories**><!-- 代码库 -->  
    <**repository**>  
        <**id**>central</**id**>  
        <**url**>http://maven.aliyun.com/nexus/content/groups/public//</**url**>  
        <**releases**>  
            <**enabled**>true</**enabled**>  
        </**releases**>  
        <**snapshots**>  
            <**enabled**>true</**enabled**>  
            <**updatePolicy**>always</**updatePolicy**>  
            <**checksumPolicy**>fail</**checksumPolicy**>  
        </**snapshots**>  
    </**repository**>  
</**repositories**>  
  
<**dependencies**>  
    <!-- kafka客户端工具 -->  
    <**dependency**>  
        <**groupId**>org.apache.kafka</**groupId**>  
        <**artifactId**>kafka-clients</**artifactId**>  
        <**version**>2.4.1</**version**>  
    </**dependency**>  
  
    <!-- 工具类 -->  
    <**dependency**>  
        <**groupId**>org.apache.commons</**groupId**>  
        <**artifactId**>commons-io</**artifactId**>  
        <**version**>1.3.2</**version**>  
    </**dependency**>  
  
    <!-- SLF桥接LOG4J日志 -->  
    <**dependency**>  
        <**groupId**>org.slf4j</**groupId**>  
        <**artifactId**>slf4j-log4j12</**artifactId**>  
        <**version**>1.7.6</**version**>  
    </**dependency**>  
  
    <!-- SLOG4J日志 -->  
    <**dependency**>  
        <**groupId**>log4j</**groupId**>  
        <**artifactId**>log4j</**artifactId**>  
        <**version**>1.2.16</**version**>  
    </**dependency**>  
</**dependencies**>  
  
<**build**>  
    <**plugins**>  
        <**plugin**>  
            <**groupId**>org.apache.maven.plugins</**groupId**>  
            <**artifactId**>maven-compiler-plugin</**artifactId**>  
            <**version**>3.7.0</**version**>  
            <**configuration**>  
                <**source**>1.8</**source**>  
                <**target**>1.8</**target**>  
            </**configuration**>  
        </**plugin**>  
    </**plugins**>  
</**build**> 

将log4j.properties配置文件放入到resources文件夹中

**log4j.rootLogger**=**INFO,stdout****  
****log4j.appender.stdout**=**org.apache.log4j.ConsoleAppender****  
****log4j.appender.stdout.layout**=**org.apache.log4j.PatternLayout****  
****log4j.appender.stdout.layout.ConversionPattern**= **%5p - %m%n**

**public class** KafkaProducerTest {  
    **public static void** main(String[] args) {  
        // 1. 创建用于连接Kafka的Properties配置  
        Properties props = **new** Properties();  
        props.put( **"bootstrap.servers"** , **"192.168.88.100:9092"** );  
        props.put( **"acks"** , **"all"** );  
        props.put( **"key.serializer"** , **"org.apache.kafka.common.serialization.StringSerializer"** );  
        props.put( **"value.serializer"** , **"org.apache.kafka.common.serialization.StringSerializer"** );  
  
        // 2. 创建一个生产者对象KafkaProducer  
        KafkaProducer<String, String> producer = **new** KafkaProducer<String, String>(props);  
  
        // 3. 调用send发送1-100消息到指定Topic test  
        **for**(**int** i = 0; i < 100; ++i) {  
            **try** {  
                // 获取返回值Future，该对象封装了返回值  
                Future<RecordMetadata> future = producer.send(**new** ProducerRecord<String, String>( **"test"** , **null**, i + **""** ));  
                // 调用一个Future.get()方法等待响应  
                future.get();  
            } **catch** (InterruptedException e) {  
                e.printStackTrace();  
            } **catch** (ExecutionException e) {  
                e.printStackTrace();  
            }  
        }  
  
        // 5. 关闭生产者  
        producer.close();  
    }  
}

**public class** KafkaProducerTest {  
    **public static void** main(String[] args) {  
        // 1. 创建用于连接Kafka的Properties配置  
        Properties props = **new** Properties();  
        props.put( **"bootstrap.servers"** , **"node1.itcast.cn:9092"** );  
        props.put( **"acks"** , **"all"** );  
        props.put( **"key.serializer"** , **"org.apache.kafka.common.serialization.StringSerializer"** );  
        props.put( **"value.serializer"** , **"org.apache.kafka.common.serialization.StringSerializer"** );  
  
        // 2. 创建一个生产者对象KafkaProducer  
        KafkaProducer<String, String> producer = **new** KafkaProducer<String, String>(props);  
  
        // 3. 调用send发送1-100消息到指定Topic test  
        **for**(**int** i = 0; i < 100; ++i) {  
            **try** {  
                // 获取返回值Future，该对象封装了返回值  
                Future<RecordMetadata> future = producer.send(**new** ProducerRecord<String, String>( **"test"** , **null**, i + **""** ));  
                // 调用一个Future.get()方法等待响应  
                future.get();  
            } **catch** (InterruptedException e) {  
                e.printStackTrace();  
            } **catch** (ExecutionException e) {  
                e.printStackTrace();  
            }  
        }  
  
        // 5. 关闭生产者  
        producer.close();  
    }  
}

**public class** KafkaProducerTest {  
    **public static void** main(String[] args) {  
        // 1. 创建用于连接Kafka的Properties配置  
        Properties props = **new** Properties();  
        props.put( **"bootstrap.servers"** , **"node1.itcast.cn:9092"** );  
        props.put( **"acks"** , **"all"** );  
        props.put( **"key.serializer"** , **"org.apache.kafka.common.serialization.StringSerializer"** );  
        props.put( **"value.serializer"** , **"org.apache.kafka.common.serialization.StringSerializer"** );  
  
        // 2. 创建一个生产者对象KafkaProducer  
        KafkaProducer<String, String> producer = **new** KafkaProducer<String, String>(props);  
  
        // 3. 调用send发送1-100消息到指定Topic test  
        **for**(**int** i = 0; i < 100; ++i) {  
            // 一、同步方式  
            // 获取返回值Future，该对象封装了返回值  
            // Future<RecordMetadata> future = producer.send(new ProducerRecord<String, String>("test", null, i + ""));  
            // 调用一个Future.get()方法等待响应  
            // future.get();  
  
            // 二、带回调函数异步方式  
            producer.send(**new** ProducerRecord<String, String>( **"test"** , **null**, i + **""** ), **new** Callback() {  
                @Override  
                **public void** onCompletion(RecordMetadata metadata, Exception exception) {  
                    **if**(exception != **null**) {  
                        System.***out***.println( **"** **发送消息出现异常** **"** );  
                    }  
                    **else** {  
                        String topic = metadata.topic();  
                        **int** partition = metadata.partition();  
                        **long** offset = metadata.offset();  
  
                        System.***out***.println( **"** **发送消息到** **Kafka** **中的名字为** **"** + topic + **"** **的主题，第** **"** + partition + **"** **分区，第** **"** + offset + **"** **条数据成功** **!"** );  
                    }  
                }  
            });  
        }  
  
        // 5. 关闭生产者  
        producer.close();  
    }  
}

ZK用来管理和协调broker，并且存储了Kafka的元数据（例如：有多少topic、partition、consumer）

ZK服务主要用于通知生产者和消费者Kafka集群中有新的broker加入、或者Kafka集群中出现故障的broker。

Kafka正在逐步想办法将ZooKeeper剥离，维护两套集群成本较高，社区提出KIP-500就是要替换掉ZooKeeper的依赖。“Kafka on Kafka”——Kafka自己来管理自己的元数据

1.1.1  producer（生产者）

生产者负责将数据推送给broker的topic

1.1.2  consumer（消费者）

消费者负责从broker的topic中拉取数据，并自己进行处理

1.1.3  consumer group（消费者组）

// 3. 发送1-100数字到Kafka的test主题中  
**while**(**true**) {  
    **for** (**int** i = 1; i <= 100; ++i) {  
        // 注意：send方法是一个异步方法，它会将要发送的数据放入到一个buffer中，然后立即返回  
        // 这样可以让消息发送变得更高效  
        producer.send(**new** ProducerRecord<>( **"test"** , i + **""** ));  
    }  
    Thread.*sleep*(3000);  
}

*// 1. 创建消费者*

    **public** **static** Consumer **<** String, String **>**  createConsumer() {

         *// 1. 创建Kafka消费者配置*

        Properties **props** **=** **new** Properties();

        **props**.setProperty( **"** **bootstrap.servers** **"** ,  **"** **node1.itcast.cn:9092** **"** );

        **props**.setProperty( **"** **group.id** **"** ,  **"** **ods_user** **"** );

        **props**.put( **"** **isolation.level** **"** , **"** **read_committed** **"** );

        **props**.setProperty( **"** **enable.auto.commit** **"** ,  **"** **false** **"** );

        **props**.setProperty( **"** **key.deserializer** **"** ,  **"** **org.apache.kafka.common.serialization.StringDeserializer** **"** );

        **props**.setProperty( **"** **value.deserializer** **"** ,  **"** **org.apache.kafka.common.serialization.StringDeserializer** **"** );

 

         *// 2. 创建Kafka消费者*

        KafkaConsumer<String, String> **consumer** **=** **new** KafkaConsumer<>(props);

 

         *// 3. 订阅要消费的主题*

        **consumer**.subscribe(**Arrays**.asList( **"** **ods_user** **"** ));

        

        **return** consumer;

}

编写一个方法 createProducer，返回一个生产者对象。注意：需要配置事务的id，开启了事务会默认开启幂等性。

1.1.1.1  编写代码消费并生产数据

实现步骤：

1. 调用之前实现的方法，创建消费者、生产者对象

2. 生产者调用initTransactions初始化事务

3. 编写一个while死循环，在while循环中不断拉取数据，进行处理后，再写入到指定的topic

(1) 生产者开启事务

(2) 消费者拉取消息

(3) 遍历拉取到的消息，并进行预处理（将1转换为男，0转换为女）

(4) 生产消息到dwd_user topic中

(5) 提交偏移量到事务中

(6) 提交事务

(7) 捕获异常，如果出现异常，则取消事务

**public** **static** void main(String[] args) {

        Consumer<String, String> **consumer** **=** createConsumer();

        Producer<String, String> **producer** **=** createProducer();

         *// 初始化事务*

        **producer**.initTransactions();

 

        **while**(true) {

            **try** {

                 *// 1. 开启事务*

                **producer**.beginTransaction();

                 *// 2. 定义Map结构，用于保存分区对应的offset*

                Map<TopicPartition, OffsetAndMetadata> **offsetCommits** **=** **new** HashMap<>();

                 *// 2. 拉取消息*

                ConsumerRecords<String, String> **records** **=** **consumer**.poll(**Duration**.ofSeconds(2));

                **for** (ConsumerRecord<String, String> **record**  **:**  records) {

                     *// 3. 保存偏移量*

                    **offsetCommits**.put(**new** TopicPartition(**record**.topic(), **record**.partition()),

                            **new** OffsetAndMetadata(**record**.offset() + 1));

                     *// 4. 进行转换处理*

                    String[] **fields** **=** **record**.value().split( **"** **,** **"** );

                    fields[1] **=** fields[1].equalsIgnoreCase( **"** **1** **"** )  **?**   **"** **男** **"** **:** **"** **女** **"** ;

                    String **message** **=** fields[0]  **+**   **"** **,** **"**   **+**  fields[1]  **+**   **"** **,** **"**   **+**  fields[2];

                     *// 5. 生产消息到dwd_user*

                    **producer**.send(**new** ProducerRecord<>( **"** **dwd_user** **"** , message));

                }

                 *// 6. 提交偏移量到事务*

                **producer**.sendOffsetsToTransaction(offsetCommits,  **"** **ods_user** **"** );

                 *// 7. 提交事务*

                **producer**.commitTransaction();

            } **catch** (Exception **e**) {

                 *// 8. 放弃事务*

                **producer**.abortTransaction();

            }

        }

    }

1.  分区和副本机制

1.1  生产者分区写入策略

生产者写入消息到topic，Kafka将依据不同的策略将数据分配到不同的分区中

1. 轮询分区策略

2. 随机分区策略

3. 按key分区分配策略

4. 自定义分区策略

1.1.1  轮询策略

加群联系作者vx：xiaoda0423

仓库地址：https://github.com/webVueBlog/JavaGuideInterview