SpringBoot整合kafka

发表于

人生有涯，学海无涯

Kafka简介

Kafka 是由 Linkedin 公司开发的，它是一个分布式的，支持多分区、多副本，基于 Zookeeper 的分布式消息流平台，它同时也是一款开源的基于发布订阅模式的消息引擎系统。

Kafka 的基本术语

  • 消息：Kafka 中的数据单元被称为消息，也被称为记录，可以把它看作数据库表中某一行的记录。
  • 批次：为了提高效率， 消息会分批次写入 Kafka，批次就代指的是一组消息。
  • 主题：消息的种类称为 主题（Topic）,可以说一个主题代表了一类消息。相当于是对消息进行分类。主题就像是数据库中的表。
  • 分区：主题可以被分为若干个分区（partition），同一个主题中的分区可以不在一个机器上，有可能会部署在多个机器上，由此来实现 kafka 的伸缩性，单一主题中的分区有序，但是无法保证主题中所有的分区有序

在这里插入图片描述

  • 生产者： 向主题发布消息的客户端应用程序称为生产者（Producer），生产者用于持续不断的向某个主题发送消息。
  • 消费者：订阅主题消息的客户端程序称为消费者（Consumer），消费者用于处理生产者产生的消息。
  • 消费者群组：生产者与消费者的关系就如同餐厅中的厨师和顾客之间的关系一样，一个厨师对应多个顾客，也就是一个生产者对应多个消费者，消费者群组（Consumer Group）指的就是由一个或多个消费者组成的群体。

在这里插入图片描述

  • 偏移量：偏移量（Consumer Offset）是一种元数据，它是一个不断递增的整数值，用来记录消费者发生重平衡时的位置，以便用来恢复数据。

broker: 一个独立的 Kafka 服务器就被称为 broker，broker 接收来自生产者的消息，为消息设置偏移量，并提交消息到磁盘保存。

broker 集群：broker 是集群 的组成部分，broker 集群由一个或多个 broker 组成，每个集群都有一个 broker 同时充当了集群控制器的角色（自动从集群的活跃成员中选举出来）。

  • 副本：Kafka 中消息的备份又叫做 副本（Replica），副本的数量是可以配置的，Kafka 定义了两类副本：领导者副本（Leader Replica） 和 追随者副本（Follower Replica），前者对外提供服务，后者只是被动跟随。
  • 重平衡：Rebalance。消费者组内某个消费者实例挂掉后，其他消费者实例自动重新分配订阅主题分区的过程。Rebalance 是 Kafka 消费者端实现高可用的重要手段。

Kafka 的特性（设计原则）

  • 高吞吐、低延迟：kakfa 最大的特点就是收发消息非常快，kafka 每秒可以处理几十万条消息，它的最低延迟只有几毫秒。
  • 高伸缩性： 每个主题(topic) 包含多个分区(partition)，主题中的分区可以分布在不同的主机(broker)中。
  • 持久性、可靠性： Kafka 能够允许数据的持久化存储，消息被持久化到磁盘，并支持数据备份防止数据丢失，Kafka 底层的数据存储是基于 Zookeeper 存储的，Zookeeper 我们知道它的数据能够持久存储。
  • 容错性： 允许集群中的节点失败，某个节点宕机，Kafka 集群能够正常工作
  • 高并发： 支持数千个客户端同时读写

Kafka 的使用场景

  • 活动跟踪：Kafka 可以用来跟踪用户行为，比如我们经常回去淘宝购物，你打开淘宝的那一刻，你的登陆信息，登陆次数都会作为消息传输到 Kafka ，当你浏览购物的时候，你的浏览信息，你的搜索指数，你的购物爱好都会作为一个个消息传递给 Kafka ，这样就可以生成报告，可以做智能推荐，购买喜好等。
  • 传递消息：Kafka 另外一个基本用途是传递消息，应用程序向用户发送通知就是通过传递消息来实现的，这些应用组件可以生成消息，而不需要关心消息的格式，也不需要关心消息是如何发送的。
  • 度量指标：Kafka也经常用来记录运营监控数据。包括收集各种分布式应用的数据，生产各种操作的集中反馈，比如报警和报告。
  • 日志记录：Kafka 的基本概念来源于提交日志，比如我们可以把数据库的更新发送到 Kafka 上，用来记录数据库的更新时间，通过kafka以统一接口服务的方式开放给各种consumer，例如hadoop、Hbase、Solr等。
  • 流式处理：流式处理是有一个能够提供多种应用程序的领域。
  • 限流削峰：Kafka 多用于互联网领域某一时刻请求特别多的情况下，可以把请求写入Kafka 中，避免直接请求后端程序导致服务崩溃。

Kafka 的消息队列

  • Kafka 是支持消费者群组的，也就是说 Kafka 中会有一个或者多个消费者，如果一个生产者生产的消息由一个消费者进行消费的话，那么这种模式就是点对点模式

在这里插入图片描述

  • 如果一个生产者或者多个生产者产生的消息能够被多个消费者同时消费的情况，这样的消息队列成为发布订阅模式的消息队列

在这里插入图片描述

Kafka 系统架构

在这里插入图片描述

如上图所示，一个典型的 Kafka 集群中包含若干Producer（可以是web前端产生的Page View，或者是服务器日志，系统CPU、Memory等），若干broker（Kafka支持水平扩展，一般broker数量越多，集群吞吐率越高），若干Consumer Group，以及一个Zookeeper集群。Kafka通过Zookeeper管理集群配置，选举leader，以及在Consumer Group发生变化时进行rebalance。Producer使用push模式将消息发布到broker，Consumer使用pull模式从broker订阅并消费消息。

核心 API

Kafka 有四个核心API，它们分别是：

  • Producer API，它允许应用程序向一个或多个 topics 上发送消息记录
  • Consumer API，允许应用程序订阅一个或多个 topics 并处理为其生成的记录流
  • Streams API，它允许应用程序作为流处理器，从一个或多个主题中消费输入流并为其生成输出流，有效的将输入流转换为输出流。
  • Connector API，它允许构建和运行将 Kafka 主题连接到现有应用程序或数据系统的可用生产者和消费者。例如，关系数据库的连接器可能会捕获对表的所有更改

在这里插入图片描述

springboot集成kafka

本地kafka和zk环境搭建

我们需要在本地启动一个单机版的kafka和zookeeper环境。kafka的安装包自带zookeeper，直接启动即可。

我的本地环境配置如下：

  • win10系统
  • kafka_2.12-2.5.0
  • zookeeper-3.4.12
  • spring boot 2.2.10.RELEASE

zookeeper的安装不做说明了，直接启动zk，端口是2181

启动kafka，首先修改配置文件config文件下的server.properties文件

listeners=PLAINTEXT://127.0.0.1:9092

log.dirs=/kafka-logs 日志存储的文件夹

到kafka根目录下，输入命令 .\bin\windows\kafka-server-start.bat .\config\server.properties ，回车；启动成功的窗口。

依赖：

1
2
3
4
java复制代码<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

application.yml

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
java复制代码server:
  port: 8999
  contextPath : /kafka
spring:
  application:
    name: kafka
  kafka:
    bootstrapServers: 127.0.0.1:9092
    consumer:
      groupId: myGroup
      keyDeserializer: org.apache.kafka.common.serialization.StringDeserializer
      valueDserializer: org.apache.kafka.common.serialization.StringDeserializer
    producer:
      groupId: myGroup
      keyDeserializer: org.apache.kafka.common.serialization.StringSerializer
      valueDserializer: org.apache.kafka.common.serialization.StringSerializer

生产者

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
java复制代码package com.cn.producer;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFutureCallback;

/**
 * 类的功能描述：
 * 消息生产者用于发送消息
 */
@Service
public class MessageProducer {

    private final Logger log = LoggerFactory.getLogger(MessageProducer.class);

    @Autowired
    KafkaTemplate kafkaTemplate;

    public void send(String payMessage) {
        kafkaTemplate.send("payTopic", payMessage).addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onFailure(Throwable ex) {
                log.info("发送消息失败："+ex.getMessage());
            }

            @Override
            public void onSuccess(SendResult<String, Object> result) {
                log.info("发送消息成功：" + result.getRecordMetadata().topic() + "-"
                        + result.getRecordMetadata().partition() + "-" + result.getRecordMetadata().offset());
            }
        });
    }
}

消费者

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
java复制代码package com.cn.consumer;

import com.cn.constant.TopicConst;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

/**
 * 类的功能描述：
 * 消息消费者者用于处理消息
 */
@Service
public class MessageConsumer {

    private final Logger logger = LoggerFactory.getLogger(MessageConsumer.class);

    @KafkaListener(topics = "payTopic")
    public void onMessage(ConsumerRecord<?, ?> record) {
        logger.info("消费者接收到消息为:"+record.topic()+"-"+record.partition()+"-"+record.value());
    }
}
  • 只需要在监听的方法上通过注解配置一个监听器即可，另外就是指定需要监听的topic。
  • kafka的消息再接收端会被封装成ConsumerRecord对象返回，它内部的value属性就是实际的消息。

发送消息

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
java复制代码package com.cn.controller;

import com.cn.producer.MessageProducer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class SendController {

    @Autowired
    private MessageProducer producer;

    @GetMapping(value = "/sendMsg/{message}")
    public void sendMsg(@PathVariable("message") String msg){
        producer.send(msg);
    }
}

测试


控制台内容如下：

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

0%