轻松搞定Kafka环境搭建和收发消息，从小白到大神！

环境准备

要使用Apache Kafka，首先需要确保机器上安装了Java。因为Kafka是基于Java构建的，因此运行它需要JDK支持。

安装JDK

1. 访问Oracle官网下载适合操作系统的JDK版本。
2. 按照官方指南完成安装与配置环境变量。

下载并启动Kafka服务

1. 从Apache Kafka官方网站或通过Maven仓库下载最新稳定版的Kafka二进制文件。
2. 解压文件后，进入bin目录。使用命令kafka-server-start.sh config/server.properties（在Windows上是.bat脚本）来启动Kafka服务器。

在Spring Boot中集成Kafka

添加依赖

在项目的pom.xml或build.gradle文件中添加以下依赖：

<dependencies>
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
        <version>2.7.8</version>
    </dependency>
</dependencies>

配置Kafka

在application.properties中配置如下内容：

# Kafka服务器地址
spring.kafka.bootstrap-servers=localhost:9092
# 指定生产者和消费者组ID
spring.kafka.consumer.group-id=test-group

创建并使用主题

创建一个名为test-topic的主题，用于发送与接收消息。

使用命令行工具创建主题

打开终端窗口，输入以下命令：

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

生产者配置及发送消息

在Spring Boot应用中编写一个简单的生产者代码。

配置生产者工厂

@Configuration
public class KafkaProducerConfig {

    @Bean
    public Map<String, Object> producerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

        return props;
    }

    @Bean
    public ProducerFactory<String, String> producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }

    @Bean
    public KafkaTemplate<String, String> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }
}

发送消息

@Autowired
private KafkaTemplate<String, String> kafkaTemplate;

public void sendMessage(String message) {
    this.kafkaTemplate.send("test-topic", message);
}

// 在Controller中调用此方法发送测试信息
@RequestMapping("/send")
public String send() {
    sendMessage("Hello Kafka");
    return "Sent";
}

消费者配置及接收消息

配置消费者工厂和监听器容器工厂

@Configuration
@EnableKafka
public class KafkaConsumerConfig {

    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "test-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

        return new DefaultKafkaConsumerFactory<>(props);
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        return factory;
    }
}

接收消息

@KafkaListener(topics = "test-topic", groupId = "test-group")
public void listen(String message) {
    System.out.println("Received: " + message);
}

安全建议

• 使用安全的端口和SSL加密连接。
• 确保所有的生产者和消费者配置中使用了适当的认证信息，比如用户名和密码或证书等。

以上步骤覆盖了从零开始搭建Kafka环境到在Spring Boot应用中实现消息收发的完整流程。通过这些基础的操作，读者可以轻松地将Kafka集成进自己的项目，并利用它强大的异步处理能力来改善系统的响应性和稳定性。