Kafka学习笔记 kafka消息中间件

Kafka学习笔记

Apache Kafka是一个分布式消息发布订阅系统。它最初由LinkedIn公司基于独特的设计实现为一个分布式的提交日志系统( a distributed commit log)，之后成为Apache项目的一部分。Kafka系统快速、可扩展并且可持久化。它的分区特性，可复制和可容错都是其不错的特性
介绍Kafka前，先来回顾一下消息中间件（MQ）的知识~~

1.消息中间件介绍

消息中间件相比大家应该不陌生，一般我们也习惯称之为MQ。它其实没有很高大上，也是一个基于语言开发出的系统，例如像：RocketMQ是阿里巴巴基于Java开发的
作用有三：应用解耦、异步处理、限流削峰
在市面上，常用的MQ有：ActiveMQ，RabbitMQ，RocketMQ，Kafka
MQ详解及四大MQ比较 - 云+社区 - 腾讯云

2.消息中间件的概念

三大核心组件
1. Broker：消息服务器，作为server提供消息核心服务
2. Provider：消息提供者
3. Consumer：消息消费者
  
  文章图片
  （集群模式）
【Kafka学习笔记】两大通信模式
1. 发布-订阅模式（也是我工作中所使用的）
2. 点对点模式

Kafka的概念

Partition(分区)：前面我们提到，可以把Topic理解为一个数据集合，那么一个Topic可以分成多个Partition（区），其中每个区的消息是有序的

若你需要所有消息都是有序的，那么你最好只用一个分区。
另外partition支持消息位移读取，消息位移有消费者自身管理

文章图片
Consumer Group（消费者组）：一群消费者的集合。向Topic订阅消费消息的单位是Consumers=ConsumerGroup，只不过Group可以是一个Consumer，也可以是多个

在我的微服务项目开发中，一般把Consumer Group设置成以微服务为单位

文章图片

关于此图的详细介绍，可以参考官方翻译： https://scala.cool/2018/03/le...

Kafka的小Demo

首先第一步需要安装环境啦，我是使用 Centos + docker + Portainer（Docker UI管理工具）去安装

Kafka需要安装 Zookeeper，Kafka和 Kafka Manager（UI管理页面）
安装也比较简单，步骤可参考：docker下安装kafka和kafka-manager - 简书
在Kafka Manager里创建 Cluster（因为Kafka是天然分布式的，都是以集群为单位创建和部署的），然后可以去创建一个 Topic
1. 文章图片
2. 文章图片
配置Yaml，再分别搭建 Provider 和 Consumer

spring: # Kafka BeanConfig kafka: bootstrap-servers: 192.XXX.XXX.X:9082 # Kafka Broker address # Kafka Producer Config producer: acks: 1 retries: 3 # retry-count key-serializer: org.apache.kafka.common.serialization.StringSerializer value-serializer: org.springframework.kafka.support.serializer.JsonSerializer # Kafka Consumer Config consumer: auto-offset-reset: earliest key-deserializer: org.apache.kafka.common.serialization.StringDeserializer value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer properties: spring: json: trusted: packages: cn.xxx.kafkademo.message # Kafka Consumer Listener Config listener: missing-topics-fatal: false

@Component public class DemoProducer { @Resource private KafkaTemplate kafkaTemplate; public SendResult syncSend(Integer id) throws ExecutionException, InterruptedException { // 创建 DemoMessage 消息 DemoMessage message = new DemoMessage(); message.setId(id); // 同步发送消息 return kafkaTemplate.send(DemoMessage.TOPIC, message).get(); } }

@Component public class DemoConsumer { private Logger logger = LoggerFactory.getLogger(getClass()); @KafkaListener(topics = DemoMessage.TOPIC, groupId = "demo-consumer-group-" + DemoMessage.TOPIC) public void onMessage(DemoMessagemessage) { logger.info("[onMessage][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), message); }}
然后再模拟Provider往 Topic里发msg，此时Consumer应该可以收到msg

文章图片

Provider 发送消息 log

[testSyncSend][发送编号：[1650810788] 发送结果：[SendResult [producerRecord=ProducerRecord(topic=KAFKA_DEMO_01, partition=null, headers=RecordHeaders(headers = [RecordHeader(key = __TypeId__, value = https://www.it610.com/article/[99, 110, 46, 105, 111, 99, 111, 100, 101, 114, 46, 115, 112, 114, 105, 110, 103,98, 111, 111, 116, 46, 108, 97, 98, 48, 51, 46, 107, 97, 102, 107, 97, 100, 101, 109, 111, 46, 109, 101, 115, 115, 97, 103, 101, 46, 68, 101, 109, 111, 48, 49, 77, 101, 115, 115, 97, 103, 101])], isReadOnly = true), key=null, value=Demo01Message{id=1650810788}, timestamp=null), recordMetadata=KAFKA_DEMO_01-0@1]]] [onMessage][线程编号:21 消息内容：DemoMessage{id=1650810788}]

Consumer 消费消息 log

[onMessage][线程编号:77 消息内容：ConsumerRecord(topic = KAFKA_DEMO_01, partition = 0, leaderEpoch = 0, offset = 1, CreateTime = 1650810788610, serialized key size = -1, serialized value size = 17, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = https://www.it610.com/article/Demo01Message{id=1650810788})] [onMessage][线程编号:79 消息内容：DemoMessage{id=1650810788}]
最后看回Kafka Manager，可以看到Topic的信息

文章图片