zookeeper搭建,可以搭建集群,也可以单机(本地学习,没必要搭建zookeeper集群,单机完全够用了,主要学习的是kafka)
1. 首先官网下载zookeeper:Apache ZooKeeper
2. 下载好之后上传到centOS or 其他虚拟机
3. 解压
4. 到zookeeper的config目录下copy 一份zoo_sample.cfg,并改名为zoo.cfg
5. 配置环境变量(这一步不是必须,如果想在任何目录下可以执行zookeeper的命令,可以执行此选项,如果不执行此选项,每次执行zookeeper命令需要到zookeeper的bin目录下去执行)
- 进入/etc 目录: cd /etc
- 修改 profile 文件: vim profile, 修改后的结果如下:
改好后,wq! 保存退出,这样就可以启动zookeeper了:zkServer.sh start
kafka 集群搭建
1. 下载kafka:Apache Kafka
2. 上传centOS,解压
3. 配置kafka环境变量,如上图所示
4. 进入config目录下,copy 2份 server.properties,重命名为 server.properties1 和 server.properties2,这一步的目的是在一台虚拟机上模拟3个kafka,只是在配置文件里做区分,每个 server.properties 需要配置以下内容,分别如下:
server.properties:
server.properties1:
server.properties2:
5. 分别启动这三个配置文件:
进入config目录下,执行以下命令
kafka-server-start.sh -daemon server.properties
kafka-server-start.sh -daemon server.properties1
kafka-server-start.sh -daemon server.properties2
**** -daemon 是后台运行
不断的刷新 JPS 命令,查看kafka启动情况:
可以看到3台kafka都已经启动,检查kafka在zookeeper里的情况
在任何目录下执行命令进入到zk 客户端:
zkCli.sh
再执行以下命令,查看zookeeper下所有的文件夹
ls /
再执行以下命令,可以查看kafka broker id情况:
ls /brokers/ids
可以看到3台kafka都已经启动了
java 代码测试
java代码:
package com.tech;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.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;import java.time.Duration;
import java.util.Collections;
import java.util.Properties;public class SimpleDemo {static class MyProducer{// 这里是centOS里的ipstatic final String BOOTSTRAP_SERVERS = "192.168.116.128:9092,192.168.116.128:9093,192.168.116.128:9094";public static void main(String[] args) {Properties properties = new Properties();properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,BOOTSTRAP_SERVERS);properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class);properties.put(ProducerConfig.CLIENT_ID_CONFIG,"MY-CLIENT");KafkaProducer<String,String> producer = new KafkaProducer(properties);ProducerRecord<String,String> record = new ProducerRecord<>("test","hello xma");producer.send(record);}}static class MyConsumer{static final String BOOTSTRAP_SERVERS = "192.168.116.128:9092,192.168.116.128:9093,192.168.116.128:9094";public static void main(String[] args) {Properties properties = new Properties();properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,BOOTSTRAP_SERVERS);properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class);properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);properties.put(ConsumerConfig.GROUP_ID_CONFIG,"TEST-GROUP");KafkaConsumer<String,String> consumer = new KafkaConsumer<String, String>(properties);consumer.subscribe(Collections.singletonList("test"));while (true) {ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(3));for (ConsumerRecord<String, String> record : records) {System.out.println(record.key()+":"+record.value());}}}}
}
producer也可以在centOS里启动:
kafka-console-producer.sh --broker-list 192.168.116.128:9092,192.168.116.128:9093,192.168.116.128:9094 --topic test
***** 注意如果consumer消费消息的时候出现如下错误,需要配置centOS里的your.host.name信息:
java.net.UnknownHostException:XXX
127.0.0.1 localhost localhost.localdomain localhost4 localhost4.localdomain4
::1 localhost localhost.localdomain localhost6 localhost6.localdomain6
192.168.116.128 your.host.name
配置好之后需重启虚拟机