针对kafka_2.13版本测试过程中的一些坑

声明:这是在windows10上进行kafka_2.13demo搭建时的过程记录,提供给同学们参考。 1.jdk先要装一下。 2.先安装zookeeper,这里不赘述,贴一个链接  https://blog.csdn.net/ring300...

声明:这是在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


条评论

请先 登录 后评论
不写代码的码农
小编

篇文章

作家榜 »

  1. 小编 文章
返回顶部
部分文章转自于网络,若有侵权请联系我们删除