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

avatar
作者
筋斗云
阅读量:0

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_kafka   datasource:     url: jdbc:mysql://localhost:3306/orders?useSSL=false&serverTimezone=UTC     username: root     password: 123456     driver-class-name: com.mysql.cj.jdbc.Driver    jpa:     hibernate:       ddl-auto: update     show-sql: true     open-in-view: false     kafka:     consumer:       bootstrap-servers: 192.168.186.77:9092,192.168.186.18:9092,192.168.186.216:9092       group-id: secKill-group       auto-offset-reset: earliest       key-deserializer: org.apache.kafka.common.serialization.StringDeserializer       value-deserializer: org.apache.kafka.common.serialization.StringDeserializer     producer:       bootstrap-servers: 192.168.186.77:9092,192.168.186.18:9092,192.168.186.216:9092       key-serializer: org.apache.kafka.common.serialization.StringSerializer       value-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;      @Version     private 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 {     @Autowired     private 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 {     @Autowired     private KafkaTemplate<String, String> kafkaTemplate;      @Autowired     private ProductService productService;      // 将秒杀请求发送到 Kafka     public 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 {         @Autowired         private KafkaMessageService kafkaMessageService;          @GetMapping("/{productId}")         public ResponseEntity<?> initiateSeckill(@PathVariable String productId) {             Object o = kafkaMessageService.sendKill("product", productId);             return ResponseEntity.ok().body(o);         }     }

6.12 项目测试

 6.12.1 网页预览

6.12.2 模拟秒杀

6.12.3 秒杀结果 

7. 总结 

        通过命令行实现kafka的快速入门，并实现简单的秒杀案例，仅供学习参考。

相关阅读
  1. 如何查看和设置我的Android N设备的权限？
  2. 如何在安卓手机上自定义应用图标？
  3. 如何在安卓微信中同时听语音和查看图片？
  4. 华为哪些设备将支持升级至安卓7.0？
  5. 微信安卓软件中Q10功能有哪些独特之处？
  6. 安卓用户如何挑选最佳视频播放器应用？
  7. 安卓系统中用户文件目录的确切位置是什么？
  8. 小米4用户如何安全升级到安卓8.0系统？
  9. 如何成功将魅蓝5升级至安卓原生7.0版本？
  10. 16GB存储的安卓手机是否能满足日常需求？

广告一刻

为您即时展示最新活动产品广告消息，让您随时掌握产品活动新动态！