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的快速入门，并实现简单的秒杀案例，仅供学习参考。