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的快速入门,并实现简单的秒杀案例,仅供学习参考。
版权归原作者 LiberInfo 所有, 如有侵权,请联系我们删除。