Kafka入门使用教程_kafka使用

1. 前言

1.1. 什么是消息队列

消息队列（MQ）是消息传递中间件解决方案的一个组件，旨在支持独立的应用和服务之间的信息交换。 消息队列按发送顺序存储“消息”（由应用所创建、供其他应用使用的数据包），直到使用方应用能够处理它们为止。 这些消息安全地等待接收方应用做好准备，因此，即使网络或接收方应用出现问题，消息队列中的消息也不会丢失。

1.2. 为什么用消息队列

1.2.1. 解耦

生产者（客户端）发送消息到MQ中去，接受者（服务端）处理消息，需要消费的系统直接去MQ取消息进行消费即可而不需要和其他系统有耦合。

例如订单服务在电商订单创建后，发送扣减可用库存消息到MQ，库存系统在接受到消息后处理扣减可用库存。后续如果还需要扩展，通知其他第三方系统，只需要新增一个消费者去接收扣减信息处理即可。

1.2.2. 异步

将用户的请求数据存储到MQ之后就立即返回结果。随后，系统再对消息进行消费。

例如短信验证码业务，登录服务在接受到用户手机号之后，将手机号发送到MQ，短信服务接受消息后进行短信发送。登录服务不需要等待发送验证码完成，即可继续后续处理。

1.2.3. 削峰

先将短时间高并发产生的事务消息存储在MQ中，然后后端服务再慢慢根据自己的能力去消费这些消息。

例如电商大促时，可在订单服务前架设一层订单接收服务，只负责接收订单并将订单信息发送到MQ，订单服务根据自身消费能力来接收订单信息并创建订单。 这样就避免订单服务因过高流量而宕机。

1.3. Kafka

Kafka是最初由Linkedin公司开发，是一个分布式、支持分区的（partition）、多副本的（replica），基于zookeeper协调的分布式消息系统，它的最大的特性就是可以实时的处理大量数据以满足各种需求场景：比如基于hadoop的批处理系统、低延迟的实时系统、storm/Spark流式处理引擎，web/nginx日志、访问日志，消息服务等等，用scala语言编写，Linkedin于2010年贡献给了Apache基金会并成为顶级开源项目。

2. 安装

2.1. 官网

Apache Kafka

2.2. docker安装

version: "1"
services:
  kafka:
    image: 'bitnami/kafka:latest'
    hostname: kafka
    ports:
      - 9092:9092
      - 9093:9093
    volumes:
      - 'D:\Docker\Kafka\data:/bitnami/kafka'
    networks:
      - kafka_net
    environment:
      # KRaft settings
      - KAFKA_CFG_NODE_ID=0
      - KAFKA_CFG_PROCESS_ROLES=controller,broker
      - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=0@kafka:9093
      # Listeners
      - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093
      - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://192.168.2.51:9092
      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER
      - KAFKA_CFG_INTER_BROKER_LISTENER_NAME=PLAINTEXT
networks:
  kafka_net:
    driver: bridge

docker-compose -f .\docker-compose.yml up -d

3. 简单使用

3.1. 可视化工具

3.1.1. offset explorer

Offset Explorer

• 配置Cluster name名字,Bootrap servers地址等信息

• 可以查看kafka节点，topic主题和consumers消费者等信息。

3.1.2. 可视化工具添加消息

• 创建topic

• 修改内容类型为String（记得点Update）

• 选中Partition，点击加号，选择Add Single Message 。Key和Value都Enter Manualy，Key可不填。点击Add

• 点击绿色Retrieve按钮，刷新分区消息。

3.2. 简单应用

3.2.1. 发送消息

• 项目中添加Kafka依赖

<!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients -->
<dependency>
  <groupId>org.apache.kafka</groupId>
  <artifactId>kafka-clients</artifactId>
  <version>3.7.0</version>
</dependency>

• 构建生产者者，填写对应的配置信息，模拟发送五次消息，并打印发送的信息

Properties properties = new Properties();
//修改为对应的bootstrap.servers
properties.put("bootstrap.servers", "192.168.2.51:9092");
properties.put("acks", "all");
properties.put("retries", 0);
properties.put("batch.size", 16384);
properties.put("linger.ms", 1);
properties.put("buffer.memory", 33554432);
 
//序列化类型。
properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
Producer<String, String> producer = null;
try {
    producer = new KafkaProducer<>(properties);
    for (int i = 0; i < 5; i++) {
        String topic = "my-topic";
        String value = "My first message" + i;
        ProducerRecord<String, String> producerRecord = new ProducerRecord<>(topic, value);
        producer.send(producerRecord);
        System.out.println("Sent:" + producerRecord);
    }
} catch (Exception e) {
    e.printStackTrace();
 
} finally {
    producer.close();
}

• 效果展示

• 工具查看Kafka中的消息

3.2.2. 消费消息

• 构建消费者，填写对应的配置信息，while循环模拟消费线程，并打印消费的信息

Properties properties = new Properties();
//修改为对应的bootstrap.servers
properties.put("bootstrap.servers", "192.168.2.51:9092");
//可自行修改，消费组名称
properties.put("group.id", "zzj.kafka");
properties.put("enable.auto.commit", "true");
properties.put("auto.commit.interval.ms", "1000");
properties.put("auto.offset.reset", "earliest");
properties.put("session.timeout.ms", "30000");
//反序列化
properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);
//制定消费的topic主题
kafkaConsumer.subscribe(Arrays.asList("my-topic"));
while (true) {
    ConsumerRecords<String, String> records = kafkaConsumer.poll(100);//100是超时时间
    for (ConsumerRecord<String, String> record : records) {
        System.out.printf("offset = %d, value = %s", record.offset(), record.value());
        System.out.println();
    }
}

• 效果展示

• 工具查看消费记录，选择对应的group.id消费组。可查看offset偏移量等信息

3.3. Springboot项目应用

3.3.1. Springboot集成Kafka

• 项目中添加Kafka依赖

<dependency>
  <groupId>org.springframework.kafka</groupId>
  <artifactId>spring-kafka</artifactId>
</dependency>

• 添加Kafka生成者和消费者配置

spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer

• 模拟生产者代码，使用spring-kafak封装的KafkaTemplate

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.*;
 
@RestController
@RequestMapping("kafka")
@Slf4j
public class KafkaController {
 
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
 
    @RequestMapping(method = RequestMethod.POST, value = "/send")
    public ResponseEntity<String> send(@RequestBody String body){
        log.info("send message: {}", body);
        kafkaTemplate.send("my-springboot-topic", body);
        return new ResponseEntity<>("success", HttpStatus.OK);
    }
}

• 消费者代码，使用@KafkaListener注解配置消费方法

package com.zzj.kafkademo.demos;
 
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;
 
@Component
@Slf4j
public class KafkaConsumer {
 
    @KafkaListener(topics = {"my-springboot-topic"}, groupId = "springboot-consumer")
    public void listen(
            //接收消息体
            @Payload String data,
 
            //接收分区编号
            @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,
 
            //接收topic主题名
            @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
 
            //接收消息的时间戳
            @Header(KafkaHeaders.RECEIVED_TIMESTAMP) long ts
    ){
        log.info("data:{},partition:{},topic:{},ts:{}", data, partition, topic, ts);
    }
}

3.3.2. 效果展示

• 使用postman模拟发送消息

• 查看发送和消费记录，KafkaController中打印发送日志，KafkaConsumer打印消费日记

4. 问题总结

4.1. 系统可用性降低：

系统可用性在某种程度上说是降低了的，为什么这样说呢？在引入Kafka之后，就需要考虑Kafka的稳定性，如果突然挂了要如何处理，如何避免Kafka挂掉等问题。

4.2. 系统复杂性提高：

引入Kafka之后，需要保证消息没有被重复消费、消息是否丢失、还要保证消息传递的顺序性，消息消费错误导致的数据一致性问题等情况。