WSL2 Ubuntu安装Kafka教程

Posted 2022-08-02 周先森爱吃素

简介

在WSL Ubuntu上安装Kafka，并配置宿主机访问。在安装Kafka之前，务必配置好JDK。

安装

首先访问官网下载合适版本的源码（建议下载推荐版本），通过如下命令下载并解压（我这里先是切换到了习惯安装软件的目录）。

cd ~/opt/
wget https://www.apache.org/dyn/closer.cgi?path=/kafka/3.2.0/kafka_2.13-3.2.0.tgz
tar -zxvf kafka_2.13-3.2.0.tgz
cd kafka_2.13-3.2.0/

为了保证宿主机的访问，我们需要修改一下Kafka-server的配置文件（ vim config/server.properties），需要将下面两项解除注释并改为如下内容。

listeners = PLAINTEXT://localhost:9092
advertised.listeners=PLAINTEXT://localhost:9092

至此，安装结束。

启动

在Kafka的安装目录下，我们首先在后台开启zookeeper服务，具体命令如下。

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

然后启动kafka-server，命令如下，此时这个终端就成了一个后台监控，不能关闭否则会终止kafka服务。（当然我们还是可以选择类似上面的方式处理为后台守护进程，但是这里为了演示不这么干。）

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

主题创建

新开一个终端并切换到kafka安装目录下，使用 kafka-topics.sh 创建单分区单副本的主题testing，命令如下：

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

成功创建后，终端会输出Created topic testing.。

我们可以通过kafka-topics.sh命令来查看当前所有的主题列表，具体如下，应该可以看到testing。

bin/kafka-topics.sh --list --bootstrap-server localhost:9092

消息处理

我们可以先进行简单的模拟看看Kafka在当前环境内（WSL）是否可以正常工作，再尝试宿主机访问能否成功。打开两个终端，均切换到kafka安装目录中的bin文件夹下，一个用来做生产者，一个用来做消费者。

首先，在第一个终端，我们通过生产者命令向指定服务器的指定主题发布消息，会打开一个交互式的输入，我们输入几个测试数据，如下图。

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

然后，在另一个终端通过消费者命令读取主题的消息，显然，当发送新的消息，它立刻能收到，如下图。

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

Spring集成

保证上面的zookeeper和kafka的服务不关闭，我们利用Spring来访问Kafaka，首先需要安装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

然后，编写如下的测试代码，模拟一个生产者发送两条消息，消费者订阅tesing主题，得到消息则输出。

package com.zc.community;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.junit.jupiter.api.Test;
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;

@SpringBootTest
@ContextConfiguration(classes = CommunityApplication.class)
public class KafkaTests 

    @Autowired
    private KafkaProducer kafkaProducer;
    @Test
    public void testKafka()
        kafkaProducer.sendMessage("testing", "wo shi zc");
        kafkaProducer.sendMessage("testing", "nihao");

        try 
            Thread.sleep(1000 * 10);
         catch (InterruptedException e)
            e.printStackTrace();
        

    


@Component
class KafkaProducer
    @Autowired
    private KafkaTemplate kafkaTemplate;

    public void sendMessage(String topic, String content)
        System.out.println("生产者发送消息:" + content);
        kafkaTemplate.send(topic, content);
    


@Component
class KafkaConsumer 
    @KafkaListener(topics = "testing")
    public void handleMessage(ConsumerRecord record)
        System.out.println("消费者收到消息:" + record.value());
    

得到的依次输出如下。

生产者发送消息:wo shi zc
生产者发送消息:nihao
消费者收到消息:wo shi zc
消费者收到消息:nihao

这代表Kafka正常运行，并且宿主机可以通过Spring集成对齐进行操作。

总结

简单介绍在WSL中Kafka的安装和使用。