博主介绍： ✌博主从事应用安全和大数据领域，有8年研发经验，5年面试官经验，Java技术专家✌
Java知识图谱点击链接：体系化学习Java（Java面试专题）
💕💕 感兴趣的同学可以收藏关注下 ，不然下次找不到哟💕💕
1、什么是 Kafka
2、Kafka 安装教程
安装前linux上需要安装jdk，这个步骤不做赘述了。并且还要安装一个 Zookeeper。
Kafka的下载地址在官网上可以找到：https://kafka.apache.org/downloads
在该网页上可以找到各个版本的Kafka安装包下载链接，包括源码和二进制包。建议选择最新版本的二进制包进行下载。
linux 在线下载使用wget命令下载Kafka安装包：
wget https://downloads.apache.org/kafka/3.5.0/kafka_2.12-3.5.0.tgz
下载日志如下：
[root@ecs-32f7 software]# wget https://downloads.apache.org/kafka/3.5.0/kafka_2.12-3.5.0.tgz
--2023-06-26 16:59:00-- https://downloads.apache.org/kafka/3.5.0/kafka_2.12-3.5.0.tgz
Resolving downloads.apache.org (downloads.apache.org)... 88.99.95.219, 135.181.214.104, 2a01:4f9:3a:2c57::2, ...
Connecting to downloads.apache.org (downloads.apache.org)|88.99.95.219|:443... connected.
HTTP request sent, awaiting response... 200 OK
Length: 107000763 (102M) [application/x-gzip]
Saving to: ‘kafka_2.12-3.5.0.tgz’
100%[===================================================================================================================================>] 107,000,763 1.90MB/s in 4m 26s
2023-06-26 17:03:31 (393 KB/s) - ‘kafka_2.12-3.5.0.tgz’ saved [107000763/107000763]
[root@ecs-32f7 software]#
接下安装包：
tar -zxvf /root/software/kafka_2.12-3.5.0.tgz
修改配置文件
vim /root/software/kafka_2.12-3.5.0/config/server.properties
broker.id=0
listeners=PLAINTEXT://localhost:9092
log.dirs=/root/software/kafka_2.12-3.5.0/kafka-logs
启动
cd /root/software/kafka_2.12-3.5.0
bin/kafka-server-start.sh config/server.properties &
3、Kafka 常用的几个命令介绍
创建 topic
./kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
使用Kafka的命令行工具 kafka-console-producer.sh 可以向Kafka发送消息。执行以下命令启动消息生产者：
./kafka-console-producer.sh --broker-list localhost:9092 --topic test
使用Kafka的命令行工具 kafka-console-consumer.sh 可以接收Kafka中的消息。执行以下命令启动消息消费者：
./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning
查看消费者组 lag 的信息
./kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group <group_name>
看某个消费者的详细信息，可以执行以下命令：
./kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group <group_name> --members --verbose
4、Kafka 的应用场景
Kafka是一个高性能、可扩展、分布式的消息队列系统，常用于以下场景：
- 数据收集与处理：Kafka可以作为数据收集和处理的中间件，用于收集和传输大量的数据，同时支持数据流处理和批处理等多种模式，非常适合大数据场景下的数据处理和分析。
- 消息系统：Kafka可以作为消息系统，用于支持实时的消息传递和处理，比如实时日志处理、实时监控和告警等场景。
- 数据存储：Kafka提供了高可靠性的数据存储机制，可以用于存储各种类型的数据，比如日志、事件、消息等，同时支持数据的持久化和复制，非常适合高可靠性和高可用性的数据存储场景。
- 流处理：Kafka提供了流处理API，可以用于实时处理数据流，支持流与流之间的连接和数据转换，非常适合实时数据处理和分析场景。
- 消息队列：Kafka本身就是一个消息队列系统，可以用于支持各种类型的消息队列应用，比如任务队列、通知队列、消息推送等场景。
总之，Kafka具有很广泛的应用场景，尤其是在大数据、实时计算和分布式系统等领域有着广泛的应用。
5、Kafka 在 java 上的基础应用代码
生产者代码如下：
package com.pany.camp.kafka;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;
/**
*
* @description: Kafka 生产者
* @copyright: @Copyright (c) 2022
* @company: Aiocloud
* @author: pany
* @version: 1.0.0
* @createTime: 2023-06-26 18:06
*/
public class KafkaProducerExample {
private static final String TOPIC_NAME = "test-topic";
private static final String BOOTSTRAP_SERVERS = "localhost:9092";
public static void main(String[] args) {
Properties props = new Properties();
props.put("bootstrap.servers", BOOTSTRAP_SERVERS);
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
for (int i = 0; i < 10; i++) {
String message = "Hello, Kafka! This is message " + i;
ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC_NAME, message);
producer.send(record);
}
producer.close();
}
}
消费者代码如下：
package com.pany.camp.kafka;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
/**
*
* @description: Kafka 消费者
* @copyright: @Copyright (c) 2022
* @company: Aiocloud
* @author: pany
* @version: 1.0.0
* @createTime: 2023-06-26 18:04
*/
public class KafkaConsumerExample {
private static final String TOPIC_NAME = "test-topic";
private static final String BOOTSTRAP_SERVERS = "localhost:9092";
private static final String GROUP_ID = "test-group";
public static void main(String[] args) {
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
props.put(ConsumerConfig.GROUP_ID_CONFIG, GROUP_ID);
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
Consumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList(TOPIC_NAME));
while (true) {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
records.forEach(record -> {
System.out.printf("Received message: key=%s, value=%s, partition=%d, offset=%d\n",
record.key(), record.value(), record.partition(), record.offset());
});
}
}
}
以上需要引入依赖
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>2.8.0</version>
</dependency>
6、Kafka 在 SpringBoot 上的应用代码
生产者
package com.pany.camp.kafka;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
/**
*
* @description: 生产者
* @copyright: @Copyright (c) 2022
* @company: Aiocloud
* @author: pany
* @version: 1.0.0
* @createTime: 2023-06-26 18:10
*/
@Component
public class KafkaProducer {
@Resource
private KafkaTemplate<String, String> kafkaTemplate;
public void sendMessage(String topic, String message) {
kafkaTemplate.send(topic, message);
}
}
消费者
package com.pany.camp.kafka;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
/**
*
* @description: 消费者
* @copyright: @Copyright (c) 2022
* @company: Aiocloud
* @author: pany
* @version: 1.0.0
* @createTime: 2023-06-26 18:10
*/
@Component
public class KafkaConsumer {
@KafkaListener(topics = "test_topic")
public void receiveMessage(String message) {
System.out.println("Received message: " + message);
}
}
配置文件 properties 配置
spring.kafka.bootstrap-servers=localhost:9092
下面是一个发送消息的例子
package com.pany.camp.kafka;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RestController;
/**
*
* @description: 发送消息
* @copyright: @Copyright (c) 2022
* @company: Aiocloud
* @author: pany
* @version: 1.0.0
* @createTime: 2023-06-26 18:12
*/
@RestController
public class KafkaController {
@Autowired
private KafkaProducer kafkaProducer;
@GetMapping("/send/{message}")
public String sendMessage(@PathVariable String message) {
kafkaProducer.sendMessage("test_topic", message);
return "Message sent: " + message;
}
}
💕💕 本文由激流丶创作，原创不易，感谢支持！
💕💕喜欢的话记得点赞收藏啊！
