zoukankan      html  css  js  c++  java
  • 针对kafka_2.13版本测试过程中的一些坑

    声明:这是在windows10上进行kafka_2.13demo搭建时的过程记录,提供给同学们参考。

    1.jdk先要装一下。

    2.先安装zookeeper,这里不赘述,贴一个链接  https://blog.csdn.net/ring300/article/details/80446918。记得测试一下zookeeper是否正确安装。

    3.下载安装kafka_2.13。在这里下载https://mirrors.tuna.tsinghua.edu.cn/apache/kafka/,并解压到你觉得OK的目录下。

     

     自己安装的kafka最好检查一下配置文件中的参数(server.properties)。1.zookeeper.connect=localhost:2181       2.log.dirs=D:\kafka_2.13-2.5.0\kafka-logs (后面的地址就是放置日志的地方,可以自己先在目录下新建,可以看上~上一张图)。

    4.开始启动服务。

    这里需要说明一下,不想cmd到文件目录下的话,请在需要打开运行窗口的地方按住 shift 然后右键 在弹出的窗口上选择 在此处打开powershell 。

    4.1先启动 zookeeper,在安装目录下的bin里直接点击zkserver.cmd 启动比较省事

     4.2启动kafka服务。

    在kafka的安装目录下直接通过(shift 然后右键 在弹出的窗口上选择 在此处打开powershell )打开powershell。然后输入   

    binwindowskafka-server-start.bat configserver.properties

         回车   就可以启动服务。

    4.3.创建一个topic   命名test(随意点就行)(shift 然后右键 在弹出的窗口上选择 在此处打开powershell )打开powershell,输入下面的命令 回车。

    创建topic:      binwindowskafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

    创建好之后,查看现有的topic:    binwindowskafka-topics.bat --list --zookeeper localhost:2181

     4.4生产消息和消费消息。

    打开shell 后 输入      binwindowskafka-console-producer.bat --broker-list localhost:9092 --topic test    生产消息

    打开shell后输入       binwindowskafka-console-consumer.bat --bootstrap-server localhost:9092 --topic test --from-beginning    消费消息

     (稍微等个几秒钟,有点慢)  

    到此就已经结束了整个test的 测试工作,接下来我们用java代码调一下这里的服务。

    pom :   "<dependencies></dependencies>"已经有了的话就去掉。

    <dependencies>
    <dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>0.10.2.0</version>
    </dependency>
    </dependencies>

    1.生产

    package com.test.kfserver;

    import java.util.Properties;
    import java.util.Random;

    import org.apache.kafka.clients.producer.KafkaProducer;
    import org.apache.kafka.clients.producer.ProducerConfig;
    import org.apache.kafka.clients.producer.ProducerRecord;
    import org.apache.kafka.common.serialization.StringSerializer;

    public class Producer {
    public static String topic = "duanjt_test";//定义主题

    public static void main(String[] args) throws InterruptedException {
    Properties p = new Properties();
    p.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");//kafka地址,多个地址用逗号分割
    p.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    p.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(p);

    try {
    while (true) {
    String msg = "Hello," + new Random().nextInt(100);
    ProducerRecord<String, String> record = new ProducerRecord<String, String>(topic, msg);
    kafkaProducer.send(record);
    System.out.println("消息发送成功:" + msg);
    Thread.sleep(500);
    }
    } finally {
    kafkaProducer.close();
    }

    }
    }

    2. 消费
    package com.test.kfserver;

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

    import org.apache.kafka.clients.consumer.ConsumerConfig;
    import org.apache.kafka.clients.consumer.ConsumerRecord;
    import org.apache.kafka.clients.consumer.ConsumerRecords;
    import org.apache.kafka.clients.consumer.KafkaConsumer;
    import org.apache.kafka.common.serialization.StringDeserializer;

    public class Consumer {

    public static void main(String[] args) {
    Properties p = new Properties();
    p.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
    p.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    p.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    p.put(ConsumerConfig.GROUP_ID_CONFIG, "duanjt_test");

    KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(p);
    kafkaConsumer.subscribe(Collections.singletonList(Producer.topic));// 订阅消息

    while (true) {
    ConsumerRecords<String, String> records = kafkaConsumer.poll(100);
    for (ConsumerRecord<String, String> record : records) {
    System.out.println(String.format("topic:%s,offset:%d,消息:%s", //
    record.topic(), record.offset(), record.value()));
    }
    }
    }

    }


    到这里我们的测试就算高一段落了,其实最新版的kafka里已经自带了zk 但是,如果用自己的zk ,只要是新版的kafka 就会报错

    zookeeper is not a recognized option  

    意思就是没有zookeeper这个参数   

    参考:https://www.cnblogs.com/duanjt/p/10132116.html

    
    
    
  • 相关阅读:
    jQuery Easing 动画效果扩展--使用Easing插件,让你的动画更具美感。
    JavaScript表达式--掌握最全的表达式,一切尽在掌握中,让表达不再是难事
    JavaScript的格式--从格式做起,做最严谨的工程师
    JavaScript 简介--对javascript的初识,最基础的了解
    手机web页面制作时的注意事项
    实现像淘宝一样牛的语音搜索框
    Cufon在渲染网页字体你不知道的事
    .net中单选按钮RadioButton,RadioButtonList 以及纯Html中radio的用法实例?
    使用C#把发表的时间改为几个月,几天前,几小时前,几分钟前,或几秒前
    eval解析JSON中的注意点
  • 原文地址:https://www.cnblogs.com/volatile0509/p/12944163.html
Copyright © 2011-2022 走看看