下面我会详细讲解“Kafka使用入门教程第1/2页”的完整攻略。

Kafka使用入门教程第1/2页

简介

Apache Kafka是一种高吞吐量、分布式的发布订阅消息系统。它最初由LinkedIn公司开发，之后成为了Apache软件基金会的一部分。Kafka的设计目标是通过Hadoop的并行加载机制来统一线上和离线消息处理的语义。

安装和环境配置

在进行Kafka开发之前，我们需要先进行安装和环境配置。我们需要安装Zookeeper和Kafka两个软件，以下是具体步骤：

1. 安装JDK

我们首先需要安装JDK，并将其配置好环境变量。Kafka需要Java 8及以上的版本支持。

2. 安装Zookeeper

由于Kafka依赖Zookeeper来保存集群信息、协调Broker之间的数据同步，我们需要事先安装和启动Zookeeper。

下载Zookeeper

我们可以在Zookeeper官网下载最新的Zookeeper。

启动Zookeeper

解压下载好的Zookeeper压缩包，并进入解压目录的bin目录，使用以下命令启动Zookeeper：

./zkServer.sh start

3. 安装Kafka

我们同样需要先下载Kafka，并解压到指定目录中。

下载Kafka

我们可以在Kafka官网下载最新的Kafka。

配置Kafka

接着，我们需要在Kafka的config目录下，修改以下文件：

• server.properties：Kafka的服务器配置文件，我们需要将zookeeper.connect配置项设置为Zookeeper的地址与端口号。例如：

zookeeper.connect=localhost:2181

启动Kafka

使用以下命令启动Kafka：

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

至此，我们已经成功安装和配置好了Kafka。

发送和接收消息

Kafka主要分为生产者(Producer)和消费者(Consumer)两个角色。以下是使用Java API来编写一个简单的生产者和一个简单的消费者。

1. 生产者

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 MyKafkaProducer {

    public static void main(String[] args) {

        String topicName = "test";
        String message = "Hello, Kafka!";

        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        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);

        ProducerRecord<String, String> record = new ProducerRecord<>(topicName, message);

        producer.send(record);

        producer.close();
    }
}

在该示例中，我们创建了一个生产者，并向名为test的主题发送了一条消息Hello, Kafka!。其中，bootstrap.servers配置项设置为Kafka的地址和端口号。

2. 消费者

import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.ConsumerRecord;

import java.util.Collections;
import java.util.Properties;

public class MyKafkaConsumer {

    public static void main(String[] args) {

        String topicName = "test";
        String groupId = "test-group";

        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", groupId);
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

        consumer.subscribe(Collections.singletonList(topicName));

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(1000);

            for (ConsumerRecord<String, String> record : records) {
                System.out.println("Received message: " + record.value() + " from partition "
                        + record.partition() + " with offset " + record.offset());
            }
        }

    }
}

在该示例中，我们创建了一个消费者，并订阅名为test的主题。当有新消息到达时，我们会在控制台上打印出该消息的内容、所在的分区和其在该分区中的偏移量。

以上就是使用Kafka发送和接收消息的简单示例代码。