分布式消息通信之Kafka的实现原理

电玩女神 2024-04-08 13:53 65阅读 0赞

### 消息中间件能做什么 ###

消息中间件主要解决的就是分布式系统之间消息传递的问题，它能够屏蔽各种平台以及协议之间的特性，实现应用程序之间的协同。举个非常简单的例子，就拿一个电商平台的注册功能来简单分析下，用户注册这一个服务，不单单只是insert一条数据到数据库里面就完事了，还需要发送激活邮件、发送新人红包或者积分、发送营销短信等一系列操作。假如说这里面的每一个操作，都需要消耗1s，那么整个注册过程就需要耗时4s才能响应给用户。

![c015062b60174e69bd93641dd30d2ebd.png][]

但是我们从注册这个服务可以看到，每一个子操作都是相对独立的，同时，基于领域划分以后，发送激活邮件、发送营销短信、赠送积分及红包都属于不同的子域。所以我们可以对这些子操作进行来实现异步化执行，类似于多线程并行处理的概念。  
如何实现异步化呢？用多线程能实现吗？多线程当然可以实现，只是，消息的持久化、消息的重发这些条件，多线程并不能满足。所以需要借助一些开源中间件来解决。而分布式消息队列就是一个非常好的解决办法，引入分布式消息队列以后，架构图就变成这样了（下图是异步消息队列的场景）。通过引入分布式队列，就能够大大提升程序的处理效率，并且还解决了各个模块之间的耦合问题。

Ø 这个是分布式消息队列的第一个解决场景【异步处理】

![d2ad2feee94843d5bdea0724eca458e7.png][]

我们再来展开一种场景，通过分布式消息队列来实现流量整形，比如在电商平台的秒杀场景下，流量会非常大。通过消息队列的方式可以很好的缓解高流量的问题。

![77227752b2024795b14289192805fbae.png][]

Ø 用户提交过来的请求，先写入到消息队列。消息队列是有长度的，如果消息队列长度超过指定长度，直接抛弃  
Ø 秒杀的具体核心处理业务，接收消息队列中消息进行处理，这里的消息处理能力取决于消费端本身的吞吐量

当然，消息中间件还有更多应用场景，比如在弱一致性事务模型中，可以采用分布式消息队列的实现最大能力通知方式来实现数据的最终一致性等等

### Java中使用kafka进行通信 ###

#### 依赖 ####

<dependency> 
         <groupId>org.apache.kafka</groupId>
         <artifactId>kafka-clients</artifactId> 
         <version>2.0.0</version> 
     </dependency>
    复制代码

#### 发送端代码 ####

public class Producer extends Thread {
            private final KafkaProducer<Integer, String> producer;
            private final String topic;
    
            public Producer(String topic) {
                Properties properties = new Properties();
                properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.13.102:9092,192 .168.13.103:9092,192.168.13.104:9092");
                properties.put(ProducerConfig.CLIENT_ID_CONFIG, "practice-producer");
                properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class.getName());
                消费端代码
                properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
                producer = new KafkaProducer<Integer, String>(properties);
                this.topic = topic;
            }
    
            @Override
            public void run() {
                int num = 0;
                while (num < 50) {
                    String msg = "pratice test message:" + num;
                    try {
                        producer.send(new ProducerRecord<Integer, String>(topic, msg)).get();
                        TimeUnit.SECONDS.sleep(2);
                        num++;
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } catch (ExecutionException e) {
                        e.printStackTrace();
                    }
                }
            }
    
            public static void main(String[] args) {
                new Producer("test").start();
            }
        }
    复制代码

#### 消费端代码 ####

public class Consumer extends Thread {
            private final

[c015062b60174e69bd93641dd30d2ebd.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/755b1620dd124703ab1fa417480f8014.png
[d2ad2feee94843d5bdea0724eca458e7.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/ad5cd44e296a49e3b1083e07e2eb96db.png
[77227752b2024795b14289192805fbae.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/51771bfd98f64603b1bc0aa7f2f25eb1.png