Kafka快速入门+SpringBoot简单的秒杀案例

1. 主题相关

1.1 创建主题

kafka-topics.sh --create --bootstrap-server [服务器地址] --replication-factor [副本数] --partitions [分区数] --topic [主题名]

liber@liber-VMware-Virtual-Platform:/home/zookeeper$ docker-compose exec kafka /bin/bash #进入kafka容器
bash-5.1# kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 3 --topic liber #创建一个主题名叫liber
Created topic liber.

注：具有 1 个副本和 3 个分区

在 Kafka 中，分区是主题的子集，每个主题可以分为多个分区。每个分区都是一个独立的日志序列，可以被存储在集群中的不同服务器上。

每个分区有一个领导者副本，负责处理所有读取和写入请求。领导者副本将写入的数据同步到其他副本。除了领导者副本外，其他副本称为追随者副本。它们从领导者那里复制数据，并不直接处理客户端的读写请求。

1.2 查询主题

kafka-topics.sh --describe --bootstrap-server localhost:9092 --topic [主题名]

bash-5.1# kafka-topics.sh --describe --bootstrap-server localhost:9092 --topic liber
Topic: liber TopicId: tTzq8pWZTIekVoXT35QPWg PartitionCount: 3 ReplicationFactor: 1 Configs: segment.bytes=1073741824
Topic: liber Partition: 0 Leader: 1 Replicas: 1 Isr: 1
Topic: liber Partition: 1 Leader: 2 Replicas: 2 Isr: 2
Topic: liber Partition: 2 Leader: 3 Replicas: 3 Isr: 3
注：如果省略 --topic 参数，则列出所有主题的详细信息。

1.3 修改主题

kafka-topics.sh --alter --bootstrap-server localhost:9092 --topic [主题名] --partitions [新的分区数]

bash-5.1# kafka-topics.sh --alter --bootstrap-server localhost:9092 --topic liber --partitions 5
bash-5.1# kafka-topics.sh --describe --bootstrap-server localhost:9092 --topic liber
Topic: liber TopicId: tTzq8pWZTIekVoXT35QPWg PartitionCount: 5 ReplicationFactor: 1 Configs: segment.bytes=1073741824
Topic: liber Partition: 0 Leader: 1 Replicas: 1 Isr: 1
Topic: liber Partition: 1 Leader: 2 Replicas: 2 Isr: 2
Topic: liber Partition: 2 Leader: 3 Replicas: 3 Isr: 3
Topic: liber Partition: 3 Leader: 1 Replicas: 1 Isr: 1
Topic: liber Partition: 4 Leader: 2 Replicas: 2 Isr: 2

注：修改liber的分区数到 5

1.4 删除主题

kafka-topics.sh --delete --bootstrap-server localhost:9092 --topic [主题名]

bash-5.1# kafka-topics.sh --delete --bootstrap-server localhost:9092 --topic liber
bash-5.1# kafka-topics.sh --describe --bootstrap-server localhost:9092 --topic liber
Error while executing topic command : Topic 'liber' does not exist as expected
[2024-07-22 02:16:33,325] ERROR java.lang.IllegalArgumentException: Topic 'liber' does not exist as expected
at kafka.admin.TopicCommand$.kafka$admin$TopicCommand$$ensureTopicExists(TopicCommand.scala:542)
at kafka.admin.TopicCommand$AdminClientTopicService.describeTopic(TopicCommand.scala:317)
at kafka.admin.TopicCommand$.main(TopicCommand.scala:69)
at kafka.admin.TopicCommand.main(TopicCommand.scala)
(kafka.admin.TopicCommand$)

2. 生产者

在 Apache Kafka中，生产者（Producer）是负责将数据发送到指定Kafka主题（Topics）的客户端应用程序。生产者可以灵活地发送消息到一个或多个Kafka主题，支持各种发布模式和消息确认机制，以确保消息的可靠性和持久性。

在 Apache Kafka 的上下文中，broker地址列表指 Kafka 集群中一组或多组 broker（服务器）的地址。这些地址用于初始化生产者（producers）、消费者（consumers）、以及其他客户端连接到Kafka集群的过程。

kafka-console-producer.sh --broker-list [broker地址列表] --topic [主题名]

bash-5.1# kafka-console-producer.sh --broker-list localhost:9092 --topic liber
>This is my first event
>This is my second event

注：Ctrl-C停止生产者客户端。

3. 消费者

在 Apache Kafka中，消费者（Consumer）是从Kafka主题（Topics）中读取数据的客户端应用。消费者可以独立使用，或者作为一个消费者群组（Consumer Group）的一部分来运行。使用消费者群组可以有效地在多个消费者实例间分配主题的分区（Partitions），从而提升数据处理的并行性和效率。

kafka-console-consumer.sh --bootstrap-server [broker地址列表] --topic [主题名] [其他可选参数]

--from-beginning：如果加上这个参数，消费者将从主题的开始读取所有消息，而不是只读取新消息。
--group：指定消费者群组的ID，用于在多个消费者间共享主题的分区。

bash-5.1# kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic liber --from-beginning
This is my first event
This is my second event

注：Ctrl-C停止消费者客户端。

4. 消费者组

4.2 隐式创建组

kafka-console-consumer.sh --bootstrap-server [broker地址列表] --topic [主题名] --group [新的或现有的消费者组ID]

消费者组的创建是隐式进行的，当一个或多个消费者客户端连接到 Kafka 并订阅主题时自动完成的。每个消费者在连接时会指定一个组ID，这个组ID在所有消费者中应该是一致的，以表示他们属于同一个消费者组。

bash-5.1# kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic liber --group example_group #创建名为example_group的用户组
注：Ctrl-C停止等待。

4.1 查询消费组（所有）

kafka-consumer-groups.sh --bootstrap-server [broker地址列表] --list

bash-5.1# kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list
example_group
KMOffsetCache-cmak

4.2 查询消费组（精确）

kafka-consumer-groups.sh --bootstrap-server [broker地址列表] --describe --group [消费者组名]

bash-5.1# kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group example_group

Consumer group 'example_group' has no active members.

GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
example_group liber 0 1 1 0 - - -
example_group liber 1 0 0 0 - - -
example_group liber 2 1 1 0 - - -

bash-5.1# kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group KMOffsetCache-cmak

GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
KMOffsetCache-cmak __consumer_offsets 22 - 0 - consumer-KMOffsetCache-cmak-1-3829d91b-249f-491c-8d69-446462d60d61 /192.168.186.77 consumer-KMOffsetCache-cmak-1
KMOffsetCache-cmak __consumer_offsets 30 - 0 - consumer-KMOffsetCache-cmak-1-3829d91b-249f-491c-8d69-446462d60d61 /192.168.186.77 consumer-KMOffsetCache-cmak-1
KMOffsetCache-cmak __consumer_offsets 25 - 0 - consumer-KMOffsetCache-cmak-1-3829d91b-249f-491c-8d69-446462d60d61 /192.168.186.77 consumer-KMOffsetCache-cmak-1
KMOffsetCache-cmak __consumer_offsets 35 - 0 - consumer-KMOffsetCache-cmak-1-3829d91b-249f-491c-8d69-446462d60d61 /192.168.186.77 consumer-KMOffsetCache-cmak-1
KMOffsetCache-cmak __consumer_offsets 37 - 0 - consumer-KMOffsetCache-cmak-1-3829d91b-249f-491c-8d69-446462d60d61 /192.168.186.77 consumer-KMOffsetCache-cmak-1
KMOffsetCache-cmak __consumer_offsets 38 - 0 - consumer-KMOffsetCache-cmak-1-3829d91b-249f-491c-8d69-446462d60d61 /192.168.186.77 consumer-KMOffsetCache-cmak-1

4.3 删除消费组

kafka-consumer-groups.sh --bootstrap-server [broker地址列表] --delete --group [消费者组名]

bash-5.1# kafka-consumer-groups.sh --bootstrap-server localhost:9092 --delete --group example_group
Deletion of requested consumer groups ('example_group') was successful.

5. 部分配置（参考）

# Kafka Broker 的基本设置
broker.id=1
# 每个 Kafka broker 需要一个唯一的 ID。在 Kafka 集群中，每个节点都必须有不同的 ID。port=9092
# Kafka 服务端监听的端口，客户端通过此端口与 Kafka 通信。num.network.threads=3
# 处理网络请求的线程数，比如接受连接、接受请求、发送响应。调整此值以匹配你的服务器的网络I/O性能。num.io.threads=8
# 服务器用于读写操作的线程数。这应该与你的磁盘数量相匹配，以平衡磁盘I/O负载。socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
# Socket 发送和接收缓冲区的大小。增加这些值可以提高网络性能，但会增加内存消耗。log.dirs=/tmp/kafka-logs
# Kafka 存储消息和日志的目录。可以指定多个目录，Kafka 会平衡跨这些目录的数据。num.partitions=1
# Kafka 创建新主题时默认的分区数。分区是并行处理的基础，更多的分区意味着更高的并发。# 数据保留策略
log.retention.hours=168
# Kafka 日志文件保留的最长时间，单位为小时。超过这个时间的日志文件将被自动删除。log.segment.bytes=1073741824
# Kafka 日志段的大小。当日志文件达到这个大小时，会新建一个日志文件。log.retention.check.interval.ms=300000
# Kafka 检查日志文件是否需要删除的频率，单位为毫秒。# 副本和同步
default.replication.factor=1
# 主题的默认副本数。副本数决定了数据的冗余程度和可用性。min.insync.replicas=1
# 在认为生产请求成功之前，必须有这么多副本同步了数据。# ZooKeeper 配置
zookeeper.connect=localhost:2181
# Kafka 使用 ZooKeeper 来维护集群状态，如存储所有broker、主题等信息。此项配置ZooKeeper服务的连接信息。zookeeper.connection.timeout.ms=6000
# 连接到 ZooKeeper 的超时时间，单位为毫秒。# 日志压缩和清理
log.cleanup.policy=delete
# 日志的清理策略。"delete" 根据时间或文件大小删除日志；"compact" 根据键合并日志。# 安全性设置
listeners=PLAINTEXT://:9092
# 定义 Kafka 服务的监听地址，支持 PLAINTEXT、SSL 等多种协议。# 高级SSL和SASL配置
# ssl.keystore.location=/path/to/keystore.jks
# ssl.keystore.password=your-keystore-pass
# ssl.key.password=your-key-pass
# sasl.enabled.mechanisms=PLAIN
# sasl.mechanism.inter.broker.protocol=PLAIN
# 配置 SSL 和 SASL，用于安全的客户端和 broker 之间的通信。

参考文档

6. 简单案例（秒杀）

6.1 创建主题

bash-5.1# kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 3 --topic product
Created topic product.
bash-5.1# kafka-topics.sh --describe --bootstrap-server localhost:9092 --topic product
Topic: product TopicId: JdkFmgvOQlKBCCsCVDTo1Q PartitionCount: 3 ReplicationFactor: 1 Configs: segment.bytes=1073741824
Topic: product Partition: 0 Leader: 1 Replicas: 1 Isr: 1
Topic: product Partition: 1 Leader: 2 Replicas: 2 Isr: 2
Topic: product Partition: 2 Leader: 3 Replicas: 3 Isr: 3

6.2 项目结构

6.3 Maven依赖

 <parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>3.3.2</version><relativePath/> <!-- lookup parent from repository --></parent><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-jpa</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId></dependency><dependency><groupId>com.mysql</groupId><artifactId>mysql-connector-j</artifactId><version>8.3.0</version></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><optional>true</optional></dependency>
</dependencies>

6.4 数据库操作

create database orders;use orders;CREATE TABLE products
(product_id   BIGINT AUTO_INCREMENT PRIMARY KEY,product_name VARCHAR(255)   NOT NULL,price        DECIMAL(10, 2) NOT NULL,stock        INT            NOT NULL,description  TEXT,version      INT            NOT NULL DEFAULT 0
);
INSERT INTO products (products.product_id,product_name, price, stock, description)
VALUES (1,'大白菜', 5.99, 200, '新鲜的大白菜，来自农民的直供'),(2,'红富士苹果', 3.50, 150, '甜美多汁的红富士苹果，一箱包含20个'),(3,'五花肉', 45.00, 100, '优质五花肉，适合各种烹饪方式'),(4,'东北大米', 60.00, 300, '东北粳米，粒粒香甜，适合日常食用'),(5,'速溶咖啡', 70.00, 80, '进口速溶咖啡，简单快捷，口味纯正');

6.5 application.yml

spring:application:name: spring_kafkadatasource:url: jdbc:mysql://localhost:3306/orders?useSSL=false&serverTimezone=UTCusername: rootpassword: 123456driver-class-name: com.mysql.cj.jdbc.Driverjpa:hibernate:ddl-auto: updateshow-sql: trueopen-in-view: falsekafka:consumer:bootstrap-servers: 192.168.186.77:9092,192.168.186.18:9092,192.168.186.216:9092group-id: secKill-groupauto-offset-reset: earliestkey-deserializer: org.apache.kafka.common.serialization.StringDeserializervalue-deserializer: org.apache.kafka.common.serialization.StringDeserializerproducer:bootstrap-servers: 192.168.186.77:9092,192.168.186.18:9092,192.168.186.216:9092key-serializer: org.apache.kafka.common.serialization.StringSerializervalue-serializer: org.apache.kafka.common.serialization.StringSerializer

6.6 SpringKafkaApplication.java

package org.example;import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;@SpringBootApplication
public class SpringKafkaApplication {public static void main(String[] args) {SpringApplication.run(SpringKafkaApplication.class, args);}
}

6.7 Product.java

package org.example.entity;import jakarta.persistence.*;
import lombok.Getter;
import lombok.Setter;import java.math.BigDecimal;@Getter
@Setter
@Entity
@Table(name = "products")
public class Product {@Id@Column(name = "product_id", nullable = false)private Long id;@Column(name = "product_name", nullable = false)private String productName;@Column(name = "price", nullable = false, precision = 10, scale = 2)private BigDecimal price;@Column(name = "stock", nullable = false)private Integer stock;@Lob@Column(name = "description")private String description;@Versionprivate int version;  // 乐观锁字段
}

6.8 ProductRepository.java

package org.example.repository;import org.example.entity.Product;
import org.springframework.data.jpa.repository.JpaRepository;
import org.springframework.stereotype.Repository;@Repository
public interface ProductRepository extends JpaRepository<Product,Long> {}

6.9 ProductService.java

package org.example.service;import org.example.entity.Product;
import org.example.repository.ProductRepository;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;@Service
public class ProductService {@Autowiredprivate ProductRepository productRepository;@Transactional//检查是否还有库存public boolean attemptPurchase(Long productId, int quantity) {Product product = productRepository.findById(productId).orElse(null);if (product != null && product.getStock() >= quantity) {product.setStock(product.getStock() - quantity);productRepository.save(product);return true;}return false;}//获取全部产品public Product getProduct(Long productId) {return productRepository.findById(productId).orElse(null);}
}

6.10 KafkaMessageService.java

package org.example.service;import org.example.entity.Product;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;@Service
public class KafkaMessageService {@Autowiredprivate KafkaTemplate<String, String> kafkaTemplate;@Autowiredprivate ProductService productService;// 将秒杀请求发送到 Kafkapublic Object sendKill(String topic, String productId) {kafkaTemplate.send(topic, productId);Product product = productService.getProduct(Long.valueOf(productId));return product;}@KafkaListener(topics = "product", groupId = "secKill-group")public void receiveKillRequest(String productId) {boolean success = productService.attemptPurchase(Long.parseLong(productId), 1);if (success) {System.out.println("秒杀成功！剩余库存："+productService.getProduct(Long.valueOf(productId)).getStock());} else {System.out.println("秒杀失败！库存不足...");}}
}

6.11 killController.java

package org.example.controller;
import org.example.service.KafkaMessageService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;@RestController
@RequestMapping("/kill")
public class killController {@Autowiredprivate KafkaMessageService kafkaMessageService;@GetMapping("/{productId}")public ResponseEntity<?> initiateSeckill(@PathVariable String productId) {Object o = kafkaMessageService.sendKill("product", productId);return ResponseEntity.ok().body(o);}}