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