1.概述

Kafka是最初由Linkedin公司开发，是一个分布式、分区的、多副本的、多订阅者，基于zookeeper协调的分布式日志系统（也可以当做MQ系统），
常见可以用于web/nginx日志、访问日志，消息服务

2.消息系统介绍

一个消息系统负责将数据从一个应用传递到另外一个应用，应用只需关注于数据，无需关注数据在两个或多个应用间是如何传递的。分布式消息传
递基于可靠的消息队列，在客户端应用和消息系统之间异步传递消息。有两种主要的消息传递模式：点对点传递模式、发布-订阅模式。大部分的消
息系统选用发布-订阅模式。Kafka就是一种发布-订阅模式。

Kafka主要设计目标如下：

  • 以时间复杂度为O(1)的方式提供消息持久化能力，即使对TB级以上数据也能保证常数时间的访问性能。
  • 高吞吐率。即使在非常廉价的商用机器上也能做到单机支持每秒100K条消息的传输。
  • 支持Kafka Server间的消息分区，及分布式消费，同时保证每个partition内的消息顺序传输。
  • 同时支持离线数据处理和实时数据处理。
  • Scale out:支持在线水平扩展

image.png

3.名词解释

image.png

  • producer：发布消息的对象称之为主题生产者（Kafka topic producer）
  • topic：Kafka将消息分门别类，每一类的消息称之为一个主题（Topic）
  • consumer：订阅消息并处理发布的消息的对象称之为主题消费者（consumers）
  • broker：已发布的消息保存在一组服务器中，称之为Kafka集群。集群中的每一个服务器都是一个代理（Broker）。 消费者可以订阅一个或多个主题（topic），并从Broker拉数据，从而消费这些已发布的消息。

4.kafka入门

image.png

（1）创建kafka-demo项目，导入依赖

  1. <dependency>
  2.     <groupId>org.apache.kafka</groupId>
  3.     <artifactId>kafka-clients</artifactId>
  4. </dependency>

（2）生产者发送消息

  1. package com.heima.kafka.sample;
  3. import org.apache.kafka.clients.producer.KafkaProducer;
  4. import org.apache.kafka.clients.producer.ProducerConfig;
  5. import org.apache.kafka.clients.producer.ProducerRecord;
  7. import java.util.Properties;
  9. /**
  10.  * 生产者
  11.  */
  12. public class ProducerQuickStart {
  14.     public static void main(String[] args) {
  15.         //1.kafka的配置信息
  16.         Properties properties = new Properties();
  17.         //kafka的连接地址
  18.         properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.200.130:9092");
  19.         //发送失败，失败的重试次数
  20.         properties.put(ProducerConfig.RETRIES_CONFIG,5);
  21.         //消息key的序列化器
  22.         properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
  23.         //消息value的序列化器
  24.         properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
  26.         //2.生产者对象
  27.         KafkaProducer<String,String> producer = new KafkaProducer<String, String>(properties);
  29.         //封装发送的消息
  30.         ProducerRecord<String,String> record = new ProducerRecord<String, String>("itheima-topic","100001","hello kafka");
  32.         //3.发送消息
  33.         producer.send(record);
  35.         //4.关闭消息通道，必须关闭，否则消息发送不成功
  36.         producer.close();
  37.     }
  39. }

（3）消费者接收消息

  1. package com.heima.kafka.sample;
  3. import org.apache.kafka.clients.consumer.ConsumerConfig;
  4. import org.apache.kafka.clients.consumer.ConsumerRecord;
  5. import org.apache.kafka.clients.consumer.ConsumerRecords;
  6. import org.apache.kafka.clients.consumer.KafkaConsumer;
  8. import java.time.Duration;
  9. import java.util.Collections;
  10. import java.util.Properties;
  12. /**
  13.  * 消费者
  14.  */
  15. public class ConsumerQuickStart {
  17.     public static void main(String[] args) {
  18.         //1.添加kafka的配置信息
  19.         Properties properties = new Properties();
  20.         //kafka的连接地址
  21.         properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.200.130:9092");
  22.         //消费者组
  23.         properties.put(ConsumerConfig.GROUP_ID_CONFIG, "group2");
  24.         //消息的反序列化器
  25.         properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
  26.         properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
  28.         //2.消费者对象
  29.         KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);
  31.         //3.订阅主题
  32.         consumer.subscribe(Collections.singletonList("itheima-topic"));
  34.         //当前线程一直处于监听状态
  35.         while (true) {
  36.             //4.获取消息
  37.             ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofMillis(1000));
  38.             for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
  39.                 System.out.println(consumerRecord.key());
  40.                 System.out.println(consumerRecord.value());
  41.             }
  42.         }
  44.     }
  46. }

总结

  • 生产者发送消息，多个消费者订阅同一个主题，只能有一个消费者收到消息（一对一）
  • 生产者发送消息，多个消费者订阅同一个主题，所有消费者都能收到消息（一对多）

5.SpringBoot集成kafka

1.导入spring-kafka依赖信息

  1. <dependencies>
  2.     <dependency>
  3.         <groupId>org.springframework.boot</groupId>
  4.         <artifactId>spring-boot-starter-web</artifactId>
  5.     </dependency>
  6.     <!-- kafkfa -->
  7.     <dependency>
  8.         <groupId>org.springframework.kafka</groupId>
  9.         <artifactId>spring-kafka</artifactId>
  10.         <exclusions>
  11.             <exclusion>
  12.                 <groupId>org.apache.kafka</groupId>
  13.                 <artifactId>kafka-clients</artifactId>
  14.             </exclusion>
  15.         </exclusions>
  16.     </dependency>
  17.     <dependency>
  18.         <groupId>org.apache.kafka</groupId>
  19.         <artifactId>kafka-clients</artifactId>
  20.     </dependency>
  21.     <dependency>
  22.         <groupId>com.alibaba</groupId>
  23.         <artifactId>fastjson</artifactId>
  24.     </dependency>
  25. </dependencies>

2.在resources下创建文件application.yml

  1. server:
  2.   port: 9991
  3. spring:
  4.   application:
  5.     name: kafka-demo
  6.   kafka:
  7.     bootstrap-servers: 192.168.200.130:9092
  8.     producer:
  9.       retries: 10
  10.       key-serializer: org.apache.kafka.common.serialization.StringSerializer
  11.       value-serializer: org.apache.kafka.common.serialization.StringSerializer
  12.     consumer:
  13.       group-id: ${spring.application.name}-test
  14.       key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
  15.       value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

3.消息生产者

  1. package com.heima.kafka.controller;
  3. import org.springframework.beans.factory.annotation.Autowired;
  4. import org.springframework.kafka.core.KafkaTemplate;
  5. import org.springframework.web.bind.annotation.GetMapping;
  6. import org.springframework.web.bind.annotation.RestController;
  8. @RestController
  9. public class HelloController {
  11.     @Autowired
  12.     private KafkaTemplate<String,String> kafkaTemplate;
  14.     @GetMapping("/hello")
  15.     public String hello(){
  16.         kafkaTemplate.send("itcast-topic","黑马程序员");
  17.         return "ok";
  18.     }
  19. }

4.消息消费者

  1. package com.heima.kafka.listener;
  3. import org.springframework.kafka.annotation.KafkaListener;
  4. import org.springframework.stereotype.Component;
  5. import org.springframework.util.StringUtils;
  7. @Component
  8. public class HelloListener {
  10.     @KafkaListener(topics = "itcast-topic")
  11.     public void onMessage(String message){
  12.         if(!StringUtils.isEmpty(message)){
  13.             System.out.println(message);
  14.         }
  16.     }
  17. }

传递消息为对象

目前springboot整合后的kafka，因为序列化器是StringSerializer，这个时候如果需要传递对象可以有两种方式
方式一：可以自定义序列化器，对象类型众多，这种方式通用性不强，本章节不介绍
方式二：可以把要传递的对象进行转json字符串，接收消息后再转为对象即可，本项目采用这种方式

发送消息

  1. @GetMapping("/hello")
  2. public String hello(){
  3.     User user = new User();
  4.     user.setUsername("xiaowang");
  5.     user.setAge(18);
  7.     kafkaTemplate.send("user-topic", JSON.toJSONString(user));
  9.     return "ok";
  10. }

接收消息

  1. package com.heima.kafka.listener;
  3. import com.alibaba.fastjson.JSON;
  4. import com.heima.kafka.pojo.User;
  5. import org.springframework.kafka.annotation.KafkaListener;
  6. import org.springframework.stereotype.Component;
  7. import org.springframework.util.StringUtils;
  9. @Component
  10. public class HelloListener {
  12.     @KafkaListener(topics = "user-topic")
  13.     public void onMessage(String message){
  14.         if(!StringUtils.isEmpty(message)){
  15.             User user = JSON.parseObject(message, User.class);
  16.             System.out.println(user);
  17.         }
  19.     }
  20. }