Kafka入门之安装试用

2019-03-11  本文已影响0人  maoruiily

前言

本文主要介绍 Kafka 在 Windows8.1,CentOS Linux release 7.4.1708 (Core)(云服务器)上安装(单机安装),以及用 java 写一个 HelloWorld 项目测试Kafka是否安装配置成功。

正文

Kafka 安装很简单,只需要去 Apache 官网下载对应的压缩文件,解压即可。但是 Kafka 依赖 Zookeeper,安装 Kafka 之前需要安装 Zookeeper。

一:Windows8.1 安装 Kafka

Zookeeper 安装

不熟悉 Zookeeper 的同学可以 点击了解下

  1. Zookeeper下载地址:http://ftp.wayne.edu/apache/zookeeper/zookeeper-3.4.13/
  2. 下载完成后解压到对应的目录,我解压到 D:\Software,根目录是D:\Software\zookeeper-3.4.13
  3. 切换到 D:\Software\zookeeper-3.4.13\conf 目录,把 zoo_sample.cfg 复制一份,重命名为 zoo.cfg
  4. 编辑 zoo.cfg 文件,修改 dataDir=/Software/zookeeper-3.4.13/data
  5. 打开cmd窗口,切换到 D:\Software\zookeeper-3.4.13\bin 目录,输入:zkServer.cmd 启动 Zookeeper
运行Zookeeper

启动结果如图:

Zookeeper运行结果

可以看到 Zookeeper 已经在 2181 端口监听了

Kafka 安装
  1. Kafka下载地址:https://www.apache.org/dyn/closer.cgi?path=/kafka/2.1.1/kafka_2.12-2.1.1.tgz
  2. 下载完成后解压到对应的目录,我解压到 D:\Software,根目录是 D:\Software\kafka_2.12-2.1.1
  3. 打开cmd窗口切换到 Kafka 安装根目录,输入:"bin/windows/kafka-server-start.bat" "config/server.properties" 启动 Kafka
    运行Kafka

启动结果如图:

Kafka运行结果

可以看到,已经成功运行了,这样在 Windows 上面安装 Kafka 就到此结束了,下面我们再来启动一个 Producer 和一个 Customer 看看效果,打开两个cmd窗口都切换到 Kafka 安装根目录:

  1. 运行 "bin/windows/kafka-topics.bat" --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test 来创建一个 topic,topic 名称是 test
  2. 运行 "bin/windows/kafka-console-producer.bat" --broker-list localhost:9092 --topic test 启动 Producer 服务来 生产消息
  3. 运行 "bin/windows/kafka-console-consumer.bat" --bootstrap-server localhost:9092 --topic test --from-beginning 启动 Customer服务来 消费消息
Kafka生产者消费者

二:云服务器Centos7.4 安装 Kafka

下载参考上面Windows,Centos7.4 不需要下载 Zookeeper,只需要下载Kafka,下载完成上传到Linux云服务器,我上传到 /root/software 目录下面

  1. 解压 tar zxvf kafka_2.12-2.1.1.tgz
  2. 我准备把Kafka安装到 /usr/kafka 目录下,如果没有创建,请创建对应的目录,执行 mv kafka_2.12-2.1.1 /usr/kafka
  3. 如果需要在本地写Deom测试云服务器上面的Kafka,还需要改下config/server.properties 文件,修改如下:
    advertised.listeners=PLAINTEXT://云服务器外网IP:9092

对,这就安装完成了,就是这么简单

现在我们来启动 Kafka 试试,启动 Kafka 之前需要启动下 Zookeeper,切换到 /usr/kafka 目录下

同样我们启动一个 Producer 和一个 Customer 看看效果

启动Producer 启动Consumer

可以看到Linux上面安装也没问题了,下面我们来用Java语言写一个Demo试试

KafkaProducerTest.java

public class KafkaProducerTest implements Runnable {

    private final KafkaProducer<String, String> producer;
    private final String topic;

    public KafkaProducerTest(String topicName) {
        Properties props = new Properties();
        props.put("bootstrap.servers", BROKER_LIST);
//        props.put("acks", "all");
        props.put("retries", 0);
        props.put("batch.size", 16384);
        props.put("key.serializer", StringSerializer.class.getName());
        props.put("value.serializer", StringSerializer.class.getName());
        this.producer = new KafkaProducer<String, String>(props);
        this.topic = topicName;
    }

    @Override
    public void run() {
        int messageNo = 1;
        try {
            for (; ; ) {
                String messageStr = "hello, this is " + messageNo + " data";
                producer.send(new ProducerRecord<String, String>(topic, "Message", messageStr));
                if (messageNo % 10 == 0) {
                    System.out.println("send success " + messageNo + " data");
                    break;
                }
                messageNo++;
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            producer.close();
        }
    }

    public static void main(String args[]) {
        KafkaProducerTest test = new KafkaProducerTest(TOPIC);
        Thread thread = new Thread(test);
        thread.start();
    }

KafkaConsumerTest.java

public class KafkaProducerTest implements Runnable {

    private final KafkaProducer<String, String> producer;
    private final String topic;

    public KafkaProducerTest(String topicName) {
        Properties props = new Properties();
        props.put("bootstrap.servers", BROKER_LIST);
//        props.put("acks", "all");
        props.put("retries", 0);
        props.put("batch.size", 16384);
        props.put("key.serializer", StringSerializer.class.getName());
        props.put("value.serializer", StringSerializer.class.getName());
        this.producer = new KafkaProducer<String, String>(props);
        this.topic = topicName;
    }

    @Override
    public void run() {
        int messageNo = 1;
        try {
            for (; ; ) {
                String messageStr = "hello, this is " + messageNo + " data";
                producer.send(new ProducerRecord<String, String>(topic, "Message", messageStr));
                if (messageNo % 10 == 0) {
                    System.out.println("send success " + messageNo + " data");
                    break;
                }
                messageNo++;
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            producer.close();
        }
    }

    public static void main(String args[]) {
        KafkaProducerTest test = new KafkaProducerTest(TOPIC);
        Thread thread = new Thread(test);
        thread.start();
    }
}

    public static final String TOPIC = "mr_topic_test";
    public static final String BROKER_LIST = "云服务器IP:9092";
    public static final String GROUP_ID = "test_group1";

各位看官动动小手试试吧,有问题欢迎评论下面留言

上一篇下一篇

猜你喜欢

热点阅读