kafka的一些信息 - 无双的个人博客

什么是Kafka

Apache Kafka是一个开源的流处理平台，由LinkedIn公司开发并于2011年贡献给Apache软件基金会。它主要用于构建实时的数据管道和流应用程序。Kafka能够以高吞吐量、可扩展性和容错性的方式处理大量数据流。它在分布式系统中广泛应用，特别是在需要处理大规模实时数据的场景中。

Kafka的主要特点包括：

发布-订阅消息系统：Kafka允许生产者发送消息到主题，消费者订阅主题以接收消息。
高吞吐量：Kafka支持高速读写操作，适合处理大量数据。
持久化：Kafka将消息持久化到磁盘，保证数据不会因为系统故障而丢失。
可扩展性：Kafka可以通过增加更多的broker来横向扩展，以支持更大的数据流。
容错性：Kafka通过副本机制确保数据的高可用性和容错性。
实时处理：Kafka能够在延迟非常低的情况下处理数据流，适合需要实时处理的应用。
流处理：Kafka有一个流处理API（Kafka Streams），允许开发者在消息产生后立即进行处理。

Kafka被广泛应用于日志收集、实时监控、事件源、流式处理、网站活动跟踪等多种场景。大型互联网公司如LinkedIn、Netflix、Uber等都在使用Kafka来处理他们的实时数据流。

它为什么快？

Kafka快的原因主要归结为以下几点：

高吞吐量：Kafka设计之初就考虑了高吞吐量，能够处理大量的数据消息。
持久化：Kafka通过将消息持久化到磁盘，并支持数据的副本，确保数据不会因为系统故障而丢失，同时也提高了读写速度。
分布式系统：Kafka是一个分布式系统，可以横向扩展，通过增加更多的服务器来提高处理能力。
零拷贝技术：Kafka利用操作系统的零拷贝（zero-copy）技术来提高数据传输速度，减少了数据在用户空间和内核空间之间的拷贝次数。
批处理：Kafka支持以批的形式处理消息，这样可以减少网络请求的次数，提高整体效率。
简化的存储机制：Kafka使用简单的顺序存储文件来存储消息，这种方式对于顺序读写非常高效。

部署Kafka

部署Apache Kafka到CentOS 7的步骤大致如下。在开始之前，请确保您的系统满足最低要求，例如已安装Java环境，因为Kafka是用Scala和Java编写的。

步骤 1: 安装Java

Kafka需要Java环境。您可以通过运行以下命令来安装OpenJDK 8:

sudo yum install java-1.8.0-openjdk

安装完成后，您可以通过以下命令确认Java版本：

java -version

步骤 2: 下载和解压Kafka

从Apache Kafka的官方网站下载最新版本的Kafka。可以使用wget命令来下载：

wget http://apache.mirrors.pair.com/kafka/2.8.0/kafka_2.13-2.8.0.tgz

请注意，下载链接可能会随着版本的更新而变化，请从官方网站获取最新的下载链接。

下载完成后，解压Kafka：

tar -xzf kafka_2.13-2.8.0.tgz
cd kafka_2.13-2.8.0

步骤 3: 配置Kafka

Kafka使用config/server.properties文件作为默认配置。我们可以使用文本编辑器打开并修改它，但对于大多数情况，可以使用默认配置。

步骤 4: 启动ZooKeeper服务

Kafka使用ZooKeeper来管理集群状态和元数据。Kafka包内包含了一个简单的单节点ZooKeeper实例，可以通过以下命令启动它：

bin/zookeeper-server-start.sh config/zookeeper.properties

在生产环境中，我们应该使用独立的ZooKeeper集群。

步骤 5: 启动Kafka服务器

在另一个终端中，启动Kafka服务器：

bin/kafka-server-start.sh config/server.properties

现在Kafka服务器应该已经在运行了。

步骤 6: 创建一个Kafka主题

为了能够发送和接收消息，需要创建一个Kafka主题。使用以下命令创建一个名为“test”的主题：

bin/kafka-topics.sh --create --topic test --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1

步骤 7: 测试Kafka生产者和消费者

可以通过Kafka命令行工具来测试消息的发送和接收。

启动生产者并发送消息：

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

在命令行中输入消息，按回车发送。

在另一个终端中，启动消费者来读取消息：

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning

现在，我们应该能够在消费者终端看到生产者发送的消息。

Java示例：

以下是一个简单的Java程序示例，演示了如何使用Apache Kafka的生产者（Producer）和消费者（Consumer）。这个例子假设已经有一个运行中的Kafka服务器和一个名为“test”的主题。

首先，确保将Kafka客户端库添加到Java项目中。如果使用Maven，可以在pom.xml文件中添加以下依赖：

<dependencies>
    <!-- Kafka客户端依赖 -->
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
        <version>2.8.0</version> <!-- 版本自定义 -->
    </dependency>
</dependencies>

Kafka生产者示例（Producer）

创建一个名为SimpleProducer.java的文件，并添加以下代码：

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

public class SimpleProducer {
    public static void main(String[] args) {
        String topicName = "test"; // Kafka主题
        String msg = "Hello Kafka"; // 发送的消息

        // 设置生产者属性
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092"); // Kafka服务器地址
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        // 创建生产者
        Producer<String, String> producer = new KafkaProducer<>(props);

        // 发送消息
        producer.send(new ProducerRecord<>(topicName, msg));

        // 关闭生产者
        producer.close();

        System.out.println("消息发送完毕");
    }
}

Kafka消费者示例（Consumer）

创建一个名为SimpleConsumer.java的文件，并添加以下代码：

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.clients.consumer.ConsumerRecord;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class SimpleConsumer {
    public static void main(String[] args) {
        String topicName = "test"; // Kafka主题

        // 设置消费者属性
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092"); // Kafka服务器地址
        props.put("group.id", "test-group"); // 消费者群组ID
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); // 从最早的消息开始读取

        // 创建消费者
        Consumer<String, String> consumer = new KafkaConsumer<>(props);

        // 订阅主题
        consumer.subscribe(Collections.singletonList(topicName));

        // 持续读取消息
        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf("收到消息：offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
                }
            }
        } finally {
            consumer.close();
        }
    }
}

在运行这些示例之前，请确保我们已经完成以下步骤：

Kafka环境准备：确保Kafka服务器正在运行，并且您已经创建了一个名为“test”的主题。如果没有，请参考之前的指导创建一个新的主题。
项目设置：确保您的Java项目已经添加了Kafka客户端的依赖。如果使用的是Maven，就像前面的pom.xml配置片段那样添加依赖。如果使用的是Gradle或其他构建工具，请相应地添加Kafka客户端库。
编译代码：使用您的IDE或构建工具编译SimpleProducer.java和SimpleConsumer.java。
运行生产者：首先运行SimpleProducer类。这将发送一个消息到Kafka的“test”主题。
运行消费者：然后运行SimpleConsumer类。如果一切正常，消费者应该会从“test”主题接收到生产者发送的消息，并将其打印到控制台。

注意，在实际部署和测试时，可能需要根据你的Kafka集群配置调整代码中的服务器地址和端口号。

此外，如果您的Kafka集群启用了安全性特性（如SASL认证或SSL加密），则需要在生产者和消费者的属性中进一步配置相关的安全性参数。

一旦您的生产者和消费者都在运行，我们应该能够看到生产者发送的消息被消费者接收并打印出来。这个简单的例子演示了Kafka客户端API的基本用法，对于学习如何在Java中使用Kafka非常有帮助。