贵安新区微信网站建设seo网站诊断顾问

kafka介绍

kafka是一个分布式的消息队列系统，适合离线和在线消费，扩展性好

kafka部署

安装包获取：
链接：https://pan.baidu.com/s/1y32yvZU-CAHBtbEfnHkJzQ
提取码：y9vb
–来自百度网盘超级会员V5的分享

安装目录为：/opt/soft/下；安装3个节点

1、配置jdk

• 第一种：yum install java-1.8.0-openjdk*（通过yum安装，修改环境变量）
• 第二种：将jdk包上传到服务器目录

注意：所有服务器节点都要安装JDK

解压

tar -xvf jdk-8u102-linux-x64.tar.gz

vim /etc/profile

JAVA_HOME=/opt/soft/jdk1.8.0_102
CLASS_PATH=.:$JAVA_HOME/lib:$JAVA_HOME/lib/tools.jar:$JAVA_HOME/jre/lib
PATH=$PATH:$JAVA_HOME/bin:$JAVA_HOME/jre/bin
export JAVA_HOME CLASS_PATH PATH

生效

source /etc/profile

检查jdk是否生效状态

jps
或
javac -version

2、zookeeper

解压安装包：

tar -xvf zookeeper-3.4.8.tar.gz
mv zookeeper-3.4.8 zookeeper（重命名，看起来整洁）

配置zoo.cfg文件

cd /opt/zookeeper/conf
cp zoo_sample.cfg zoo.cfg

编辑，vim zoo.cfg

tickTime=2000 #服务器与客户端之间交互的基本时间单元（ms） 
initLimit=10  #zookeeper所能接受的客户端数量 
syncLimit=5 #服务器和客户端之间请求和应答之间的时间间隔
dataDir=/opt/soft/zookeeper/data  #保存zookeeper数据，日志的路径
quorumListenOnAllIPs=true #因为配置的结点不是在同一内网中，设置的是公网ip
server.1=192.168.64.20:2888:3888
server.2=192.168.64.21:2888:3888
server.3=192.168.64.22:2888:3888   #根据节点ip进行修改

创建数据目录

mkdir -p /opt/soft/zookeeper/data

将zookeeper目录，拷贝到各个节点目录

scp -r zookeeper 192.168.64.21:/opt/soft
scp -r zookeeper 192.168.64.22:/opt/soft

给每个节点创建myid（myid的值需要与配置文件中server的id相同）

vim /opt//softzookeeper/data/myid（20节点）
1

vim /opt//softzookeeper/data/myid（21节点）
2

vim /opt/soft/zookeeper/data/myid（22节点）
3

启动zookeeper并查看状态（各个节点）

# 启动zookeeper：
/opt/opt/zookeeper/bin/zkServer.sh start
# 查看zookeeper状态：
/opt/opt/zookeeper/bin/zkServer.sh status
#jps查看是否启动QuorumPeerMain
#停止服务
/opt/opt/zookeeper/bin/zkServer.sh stop

3、kafka安装

解压安装包

tar -xvf kafka_2.11-0.10.0.0.tgz 
mv kafka_2.11-0.10.0.0 kafka（重命名，看起来简洁）

cd kafka/config/ 目录，配置server.properties文件

vim server.properties

broker.id=0       //broker.id 需要kafka集群各节点不相同
delete.topic.enable=true   //允许管理人员删除topic
auto.create.topics.enable=false    //手动控制topic的创建，否则goldengate可以自动创建topic
listeners=PLAINTEXT://192.168.64.20:9092  //配置为本节点的ip
log.dirs=/opt/soft/kafka/logs
zookeeper.connect=192.168.64.20:2181, 192.168.64.21:2181, 192.168.64.22:2181
log.retention.hours=168 //多少个小时后，log数据会被删除，优先级低于minutes和ms
#log.retention.minutes=60 //多少分钟后，log数据会被删除，优先级低于ms
#log.retention.ms=6000  //多少毫秒后，log数据会被删除，如果被配置则优先使用
log.retention.bytes=-1  //log文件达到多少字节后，数据会被删除。设置为-1则没有大小限制
log.segment.bytes=1073741824  //每个log分片的大小，超过后会新建分片
log.roll.ms=60000            //间隔多久新创建一个数据文件
log.retention.check.interval.ms=300000  //多长时间轮询检查是否会删除log

创建日志目录

mkdir -p /opt/soft/kafka/logs

将kafka传给要安装其他各个节点

scp -r kafka 192.168.64.21:/opt/soft
scp -r kafka 192.168.64.22:/opt/soft

修改各个节点的broker.id和listeners（节点的broker.id的值不一样，分别为1，2，3）
vim /opt/kafka/config/server.properties（21节点）

broker.id=1
listeners=PLAINTEXT://192.168.64.21:9092

vim /opt/kafka/config/server.properties（22节点）

broker.id=2
listeners=PLAINTEXT://192.168.64.22:9092

启动kafka，检验是否运行（各个节点）

• 启动，jsp检验是否启动

/opt/soft/kafka/bin/kafka-server-start.sh /opt/soft/kafka/config/server.properties>/opt/kafka/logs/kk.log &/opt/soft/kafka/bin/kafka-server-start.sh  -daemon /opt/soft/kafka/config/server.properties

检测2181和9092端口

netstat -tunlp|egrep "(2181|9092)"

kafka使用测试

创建topic

/opt/soft/kafka/bin/kafka-topics.sh --create --partitions 1 --replication-factor 3 --zookeeper 192.168.64.20:2181, 192.168.64.21:2181, 192.168.64.22:2181 --topic test

检查topic

/opt/soft/kafka/bin/kafka-topics.sh --list --zookeeper 192.168.64.20:2181, 192.168.64.21:2181, 192.168.64.22:2181

打开生产者

/opt/soft/kafka/bin/kafka-console-producer.sh --broker-list 192.168.192.20:9092 --topic test

打开消费者

/opt/soft/kafka/bin/kafka-console-consumer.sh --zookeeper 192.168.64.20:2181, 192.168.64.21:2181, 192.168.64.22:2181 --topic test --from-beginning