准备三台互通服务器(关闭防火墙)
192.168.203.136
192.168.203.137
192.168.203.138
一.搭建zookeeper集群
1、在/usr/local目录下解压安装包,
tar -zxvf apache-zookeeper-3.7.1-bin.tar.gz
链接:百度网盘 请输入提取码
提取码:bd2u
2、进入zookeeper目录下新建文件夹zkData
mkdir zkData
3、进入conf目录下修改配置文件
mv zoo_sample.cfg zoo.cfg
vi zoo.cfg
(修改dataDir参数)
dataDir=/usr/local/zookeeper-3.7.1/zkData
(新增下面三行,2888端口用于主从信息同步,3888端口用于选举)
server.1=192.168.203.136:2888:3888
server.2=192.168.203.138:2888:3888
server.3=192.168.203.137:2888:3888
保存退出
4、进入zkData目录,将myid文件上传到该目录下(myid中的数字与上面server.1 2 3表示的服务器对应)
5、安装剩余的两台服务器(重复上面操作,myid要修改为对应的数字)
6、依次进入bin目录启动三台zookeeper
./zkServer.sh start (启动)
./zkServer.sh status (查看状态,一主两从即为成功)
二.搭建kafka集群
链接:百度网盘 请输入提取码
提取码:q4pm
1、在/usr/local目录下解压安装包
tar -zxvf kafka_2.12-3.3.2.tgz
2、进入kafka/config/目录下修改配置文件
vi server.properties(broker.id 不能相同,可以依次设置为0,1,2。ip改为对应服务器的ip地址)
broker.id=0(修改参数)
port=9092 (新增参数)
host.name=192.168.203.136 (新增参数)
steners=PLAINTEXT://192.168.203.136:9092 (新增参数)
advertised.listeners=PLAINTEXT://192.168.203.136:9092 (新增参数)
zookeeper.connect=192.168.203.136:2181,192.168.203.138:2181,192.168.203.137:2181(修改参数)
3、安装剩余的两台服务器(重复上面操作,修改broker.id)
4、依次进入kafka目录下,启动kafka
./bin/kafka-server-start.sh -daemon ./config/server.properties &
netstat -tunlp |grep 9092 (检查kafka端口)
注1:kafak新版本操作命令
启动命令:
bin/kafka-server-start.sh -daemon config/server.properties
创建topic
./kafka-topics.sh --create --bootstrap-server 192.168.203.136:9092 --replication-factor 1 --partitions 1 --topic test
查看topic
./kafka-topics.sh --bootstrap-server 192.168.203.136:9092 --list
向指定topic中生产数据
./kafka-console-producer.sh --broker-list 192.168.203.136:9092 --topic test
例如:{"id":"1","name":"xiaoming","age":"20"}
查看topic具体内容
./kafka-console-consumer.sh --bootstrap-server 192.168.203.136:9092 --topic test --from-beginning
创建消费者组
./kafka-console-consumer.sh --bootstrap-server 192.168.203.136:9092 --topic test --group consumer-10
查看消费者组
./kafka-consumer-groups.sh --bootstrap-server 192.168.203.136:9092 --list
查看消费者详情
./kafka-consumer-groups.sh --bootstrap-server 192.168.203.136:9092 --describe --group consumer-10
消费数据
./kafka-console-consumer.sh --bootstrap-server 192.168.203.136:9092 --topic test --from-beginning
注2:kafka集群监控系统安装
1、选择集群中任意一台服务器上传并解压安装包
unzip kafka-manager-2.0.0.2.zip
2、进入kafka-manager/conf目录下修改配置文件
vi application.conf
kafka-manager.zkhosts="192.168.203.136:2181" (修改参数ip为zookeeper主机ip)
3、返回kafka-manager目录启动系统
nohup bin/kafka-manager -Dconfig.file=conf/application.conf -Dhttp.port=9000 >kafka-manager.log 2>&1 &
浏览器访问ip:9000端口即可。
注3:java连接kafka集群:
1、pom文件增加依赖
<!-- kafka -->
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
2、修改配置文件(groupId自定义)
kafka:
bootstrapServers: 192.168.203.136:9092,192.168.203.138:9092,192.168.203.137:9092
producer:
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.apache.kafka.common.serialization.StringSerializer
consumer:
groupId: consumer-10
auto-offset-reset: latest
value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
enable-auto-commit: false
auto-commit-interval: 5000
max-poll-records: 1
3、生产者
@Autowired
private KafkaTemplate kafkaTemplate;
@Scheduled(cron = "0/5 * * * * ?")
public void kafkaTest(){
kafkaTemplate.send("test","测试kafka+时间为"+LocalDateTime.now());
log.info("发送kafka test 数据:");
}
4、消费者
package com.example.demo.controller;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
@Component
@Slf4j
public class KafkaListener1 {
@KafkaListener(topics = "test")
public void listenKafkaHour1(String payMessage) {
log.info("收到kafka test 数据:" + payMessage);
}
}
版权归原作者 请叫我滚去学习。 所有, 如有侵权,请联系我们删除。