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

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

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

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

<**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.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();  
    }  
}

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

}

**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();

            }

        }

    }