-
Kafka 简介
– Kafka是一个分布式的流媒体平台。
– 应用:消息系统、日志收集、用户行为追踪、流式处理。 -
Kafka特点
高吞吐量、消息持久化、高可靠性、高扩展性。 -
Kafka术语
– Broker（服务器）、Zookeeper
– Topic（主题）、Partition（分区）、 Offset（索引）
– Leader Replica（主副本，功能强，可以提供数据）、Follower Replica（随从副本，主要是用作备份）
我们首先深入了解 Kafka 为记录流提供的核心抽象 - 主题topics
一个Topic可以认为是一类消息，每个topic将被分成多个partition(区),每个partition在存储层面是append log文件、
主题是发布记录的类别或订阅源名称。Kafka的主题总是多用户; 也就是说，一个主题可以有零个，一个或多个消费者订阅写入它的数据。
对于每个主题，Kafka集群都维护一个如下所示的分区日志：
每个分区都是一个有序的，不可变的记录序列，不断附加到结构化的提交日志中。分区中的记录每个都分配了一个称为偏移的顺序ID号，它唯一地标识分区中的每个记录。
Kafka集群持久保存所有已发布的记录 - 无论是否已使用 - 使用可配置的保留期。例如，如果保留策略设置为两天，则在发布记录后的两天内，它可供使用，之后将被丢弃以释放空间。Kafka的性能在数据大小方面实际上是恒定的，因此长时间存储数据不是问题。
实际上，基于每个消费者保留的唯一元数据是该消费者在日志中的偏移或位置。这种偏移由消费者控制：通常消费者在读取记录时会线性地提高其偏移量，但事实上，由于该位置由消费者控制，因此它可以按照自己喜欢的任何顺序消费记录。例如，消费者可以重置为较旧的偏移量来重新处理过去的数据，或者跳到最近的记录并从“现在”开始消费。
这些功能组合意味着Kafka 消费者consumers 非常cheap - 他们可以来来往往对集群或其他消费者没有太大影响。例如，您可以使用我们的命令行工具“tail”任何主题的内容，而无需更改任何现有使用者所消耗的内容。
日志中的分区有多种用途。首先，它们允许日志扩展到超出适合单个服务器的大小。每个单独的分区必须适合托管它的服务器，但主题可能有许多分区，因此它可以处理任意数量的数据。其次，它们充当了并行性的单位 - 更多的是它。
进官网下载kafka
点击download kafka
因为kafka里面自带linux和Windows命令，所以在linux和Windows操作系统兼容
- 进入安装目录例如我的是：D:environmentkafkakafka_2.13-3.1.0config
- 打开以下文件进行修改
zookeeper.properties：
server.properties
同理，修改如下：
按win+R输入cmd
进入安装目录输入 binwindowszookeeper-server-start.bat configzookeeper.properties
结果如图：
-
零开一窗口进入目录输入binwindowskafka-server-start.bat configserver.properties
-
此时data文件夹多了两个文件
-
再另开一窗口进入目录执行执行kafka-topics.bat --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic test
-
查看正在运行的服务列表
-
另开一窗口进入目录模拟生产者
kafka-console-producer.bat --broker-list localhost:9092 --topic test
-
另开一窗口模拟消费者
kafka-console-consumer.bat -bootstrap-server localhost:9092 --topic test --from-beginning
此时回到生产者再次发送消息
消费者也同步接收到消息
- 导入依赖
org.springframework.kafka spring-kafka
- application.properties
#KafkaProperties spring.kafka.bootstrap-servers=localhost:9092 spring.kafka.consumer.group-id=test-consumer-group spring.kafka.consumer.enable-auto-commit=true spring.kafka.consumer.auto-commit-interval=3000
- 测试类
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringRunner;
@RunWith(SpringRunner.class)
@SpringBootTest
@ContextConfiguration(classes = CommunityApplication.class)
public class KafkaTests {
@Autowired
private KafkaProducer kafkaProducer;
@Test
public void testKafka() {
kafkaProducer.sendMessage("test", "你好");
kafkaProducer.sendMessage("test", "世界");
// 为了能看到消费者执行的过程，这里需要等待一下
try {
Thread.sleep(1000 * 10);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
@Component
class KafkaProducer {
@Autowired
private KafkaTemplate kafkaTemplate;
public void sendMessage(String topic, String content) {
kafkaTemplate.send(topic, content);
}
}
@Component
class KafkaConsumer {
// 消费者是自动接收消息的
@KafkaListener(topics = {"test"})
public void handleMessage(ConsumerRecord record) {
System.out.println(record.value());
}
}
结果：
