前往小程序,Get更优阅读体验!
立即前往
首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >消息队列-Kafka(1)

消息队列-Kafka(1)

作者头像
lpe234
发布2021-04-13 16:22:32
1.1K0
发布2021-04-13 16:22:32
举报
文章被收录于专栏:若是烟花

Apache Kafka is an open-source distributed event streaming platform used by thousands of companies for high-performance data pipelines, streaming analytics, data integration, and mission-critical applications.

1 概述

1.1 基本概念
1.1.1 Broker 代理

已发布的消息保存在一组服务器中,称为Kafka集群。集群中的每个服务器都是一个Broker。

1.1.2 Topic 主题

通过Topic机制对消息进行分类,可以认为每个Topic就是一个队列。

1.1.3 Partition 分区

每个Topic可以有多个分区,主要为了提高并发而设计。相同Topic下不同Partition可以并发接收消息,同时也能供消费者并发拉取消息。有多少Partition就有多少并发量。

在Kafka服务器上,分区是以文件目录的形式存在的。每个分区目录中,Kafka会按配置大小及配置周期将分区拆分成多个段文件(LogSegment),每个段由三部分组成:

代码语言:javascript
复制
- 日志文件:*.log
- 位移索引文件:*.index
- 时间索引文件:*.timeindex

其中*.log用于存储消息本身的数据内容,*.index存储消息在文件中的位置(包括消息的逻辑offset和物理存储offset),*.timeindex存储消息创建时间和对应逻辑地址的映射关系。

将分区拆分成多个段是为了控制存储文件大小。可以很方便的通过操作系统mmap机制映射到内存中,提高写入和读取效率。同时还有一个好处就是,当系统要清除过期数据时,可以直接将过期的段文件删除。

如果每个消息都要在index中保存位置信息,index文件自身大小也很容易变的很大。所以Kafka将index设计为稀疏索引来减小index文件的大小。

1.1.4 Replication 副本

消息冗余数量。不能超过集群中Broker的数量。

1.2 基本操作
1.2.1 Topic相关
代码语言:javascript
复制
# 创建Topic 
# --topic 主题名称 避免使用[_]及[.]号
# --replication-factor 副本数量(不能超过broker节点数)
# --partitions 分区数量(并发)
./bin/kafka-topics.sh --create \
--topic UserDataQueue \
--replication-factor 3 \
--partitions 5 \
--bootstrap-server localhost:9092,localhost:9093,localhost:9094

# 查看Topic
./bin/kafka-topics.sh --list \
--bootstrap-server localhost:9092,localhost:9093,localhost:9094

# 修改Topic
# 删除Topic
1.2.2 Message相关
代码语言:javascript
复制
# 发送消息
# --topic 指定目标Topic
./bin/kafka-console-producer.sh \
--topic UserDataQueue \
--bootstrap-server localhost:9092,localhost:9093,localhost:9094

# 拉取消息
# --from-beginning 从头开始(获取现有的全量数据)
./bin/kafka-console-consumer.sh \
--topic UserDataQueue \
--bootstrap-server localhost:9092,localhost:9093,localhost:9094 \
--from-beginning

2 集群配置

Kafka集群依赖于Zookeeper。

2.1 Zookeeper配置及启动
代码语言:javascript
复制
# 需要修改的参数

# the directory where the snapshot is stored.
dataDir=/kafka/zkdata
# the port at which the clients will connect
clientPort=2182
代码语言:javascript
复制
# 启动
./bin/zookeeper-server-start.sh -daemon /kafka/zookeeper.properties
2.2 Kafka配置及启动
代码语言:javascript
复制
# 需修改参数

# The id of the broker. This must be set to a unique integer for each broker.
broker.id=1  # 同一集群内ID必须唯一

# The address the socket server listens on. It will get the value returned from 
# java.net.InetAddress.getCanonicalHostName() if not configured.
#   FORMAT:
#     listeners = listener_name://host_name:port
#   EXAMPLE:
#     listeners = PLAINTEXT://your.host.name:9092
listeners=PLAINTEXT://localhost:9092  # 同一主机的话,端口号不能相同

# A comma separated list of directories under which to store log files
log.dirs=/kafka/data01  # 日志存储目录,需做隔离

# Zookeeper connection string (see zookeeper docs for details).
# This is a comma separated host:port pairs, each corresponding to a zk
# server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".
# You can also append an optional chroot string to the urls to specify the
# root directory for all kafka znodes.
zookeeper.connect=localhost:2182  # Zookeeper连接地址,参见2.1 zk配置
代码语言:javascript
复制
# Kafka启动

# broker-1
./bin/kafka-server-start.sh -daemon /kafka/server01.properties

# broker-2
./bin/kafka-server-start.sh -daemon /kafka/server02.properties

# broker-3
./bin/kafka-server-start.sh -daemon /kafka/server03.properties
2.3 Zookeeper可视化

PrettyZoo 是一个基于 Apache Curator 和 JavaFX 实现的 Zookeeper 图形化管理客户端。

由下图可以看到,集群3个Broker均正常启动。

2.4 Kafka可视化及监控
2.4.1 AKHQ

管理Topic,Topic消息,消费组等的Kafka可视化系统,相关文档:https://akhq.io/

2.4.2 Kafka Eagle

一个简单且高效的监控系统。相关文档:http://www.kafka-eagle.org/index.html

Kafka Eagle 自带监控大屏。

3 与Spring Boot集成

Spring Boot版本:2.4.4。

官方示例:https://github.com/spring-projects/spring-kafka/tree/main/samples

3.1 Spring Boot
3.1.1 添加依赖
代码语言:javascript
复制
implementation 'org.springframework.kafka:spring-kafka'
3.1.2 配置文件
代码语言:javascript
复制
spring:
  kafka:
    bootstrap-servers: localhost:9092,localhost:9093,localhost:9094
    producer:
      client-id: kfk-demo
      retries: 3
3.1.3 消息发送
代码语言:javascript
复制
@RestController
public class IndexController {

    @Autowired
    KafkaTemplate<Object, Object> kafkaTemplate;

    @GetMapping
    public String index() {
        int rdm = new Random().nextInt(1000);
        kafkaTemplate.send("UserDataQueue", new UserData("", rdm));
        return "hello world";
    }

    @GetMapping("index2")
    public String index2() {
		// 发送字符串方式
        kafkaTemplate.send("UserDataTopic", new Gson().toJson(new UserData("apple", 23)));
        return "ok";
    }
}
3.1.4 消息接收
代码语言:javascript
复制
@Component
@KafkaListener(
        id = "kfk-demo-userdata",
        topics = {"UserDataQueue"},
        groupId = "kfk-demo-group",
        clientIdPrefix = "kfk-demo-client")
public class KfkListener {

    @KafkaHandler
    public void process(@Payload UserData ud,
                        @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
                        @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition) {
        System.out.println(String.format("topic: %s, partition: %d, userData: %s", topic, partition, ud));
    }

    @KafkaHandler(isDefault = true)
    public void process(Object obj) {
        System.out.println(obj);
    }
}

// 接收字符串方式
@Slf4j
@Component
@KafkaListener(id = "kfk-demo2", topics = {"UserDataTopic"})
public class KfkUserDataTopicListener {

    @KafkaHandler
    public void process(String userDataStr) {
        UserData userData = new Gson().fromJson(userDataStr, UserData.class);
        log.info("username: {}, age: {}", userData.getUsername(), userData.getAge());
    }
}
3.1.5 Topic自动创建
代码语言:javascript
复制
@Configuration
public class KafkaConfig {

    @Bean
    public NewTopic userDataTopic() {
        return new NewTopic("UserDataTopic", 3, (short) 1);
    }
}

参考文档

本文参与 腾讯云自媒体同步曝光计划,分享自作者个人站点/博客。
如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 作者个人站点/博客 前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体同步曝光计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
目录
  • 1 概述
    • 1.1 基本概念
      • 1.1.1 Broker 代理
      • 1.1.2 Topic 主题
      • 1.1.3 Partition 分区
      • 1.1.4 Replication 副本
    • 1.2 基本操作
      • 1.2.1 Topic相关
      • 1.2.2 Message相关
  • 2 集群配置
    • 2.1 Zookeeper配置及启动
      • 2.2 Kafka配置及启动
        • 2.3 Zookeeper可视化
          • 2.4 Kafka可视化及监控
            • 2.4.1 AKHQ
            • 2.4.2 Kafka Eagle
        • 3 与Spring Boot集成
          • 3.1 Spring Boot
            • 3.1.1 添加依赖
            • 3.1.2 配置文件
            • 3.1.3 消息发送
            • 3.1.4 消息接收
            • 3.1.5 Topic自动创建
        • 参考文档
        相关产品与服务
        对象存储
        对象存储(Cloud Object Storage,COS)是由腾讯云推出的无目录层次结构、无数据格式限制,可容纳海量数据且支持 HTTP/HTTPS 协议访问的分布式存储服务。腾讯云 COS 的存储桶空间无容量上限,无需分区管理,适用于 CDN 数据分发、数据万象处理或大数据计算与分析的数据湖等多种场景。
        领券
        问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档