jdk-kafka-zookeeper都解压后配置java环境变量cd kafka/binnohup bin/zookeeper-server-start.sh config/zookeeper.properties nohup bin/kafka-server-start.sh config/server.properties jps#创建topic[rootmaster bin]# ./kafka-topics.sh --bootstrap-server localhost:9092 --create --topic test#创建生产者[rootmaster kafka]# bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic test#创建消费者[rootmaster bin]# ./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test#同一个消费者组中只会在其中一个消费一次不会重复消费bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group testgroup#查看消费者组信息总共有多少消息接受了多少多少没接收bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group testgroup#------------------------------------------------------------------------------------------------------------------------------#搭建集群vi /etc/hostsip work1ip work2ip work3#三个节点都安装zookeepercd /app/zookeepercd confcp zoo_sample.cfg zoo.cfgvi zoo.cfg#创建数据目录mkdir -p /app/zookeeper/datacd /app/zookeeper/dataecho 1 myid#20写121写222写3#启动zookeeper三个节点都搞cd /app/zookeeperbin/zkServer.sh --config conf start#三个节点也都要安装kafkacd /kafka/configvi server.properties#不同节点要不一样区分id#listen#改日志路径#连接到zookeeper集群#改完可以通过scp复制过去就再做些小修改就行scp -r /app/kafka/ rootwork2:/app/kafka#启动kafkacd /app/kafka#daemon是后台启动bin/kafka-server-start.sh -daemon config/server.propertiesjps#确保三个都启动成功了javazookeeper,kafka#创建一个topicbin/kafka-topics.sh --bootstrap-server work1:9092 --create --topic distopic --replication-factor 2 --partitions 4#开启四个partitions 两个备份#创建生产者bin/kafka-console-producer.sh --bootstrap-server work1:9092 --create --topic distopic#创建消费者并设置组bin/kafka-console-consumer.sh --bootstrap-server work1:9092 --group testgroup --topic distopic#通过组查看消息队列数据bin/kafka-consumer-group.sh --bootstrap-server wor1:9092 --describe --group testgroup#这些leader才是真实发送数据的不管你是在--bootstrap-server指定其他节点最终都是找到leader再转发同步的#初始化选举按顺序启动时第三个被选举为leader因为要选举第一个只有一个投自己没用自动变为从节点第二个双数节点没用第三个节点上来时就选举为leader了#znode#进入znodecd /app/zookeeper/bin./zkCli.sh#指定节点 -server./zkCli.sh -server 192.168.255.21#znode管理命令ls -s 路径 #查看详细信息等同于 stat 路径#类似都是操作目录文件createstatls#create加参数e指创建临时节点create -e#set设置值key value形式#每set一次会更新 datalength 是实际的value长度并且 dataversion 会加一#watcher时间 -w参数,只能单次应该是用来检测程序的stat -w /opt执行后下次修改/opt 会提示修改了什么父节点检测修改子节点不触发监控#delete只能当节点是空值时删除当节点有·子节点时再像删除要deleteall#ACL权限控制#查看权限getAcl /path#权限都是针对起直属的第一个子节点子节点之后就约束不了了假设当父节点有c权限创建了子节点那子节点默认是最大权限cdrwa#修改权限setAcl /opt world:anyone:crwa#加参数-r就可以遍历子节点设置权限,就可以让子节点全都没用删除权限setAcl -R /optworld:anyone:crwa#auth#addauth先创建一个用户密码——digest就是对密码加密addauth digest ljh:123456#在有这个账号的基础上就可以给一些文件设置账号密码登录原账号是(worldanyone)#给/z1设置用户密码setAcl auth:ljh:123456:crwad之后在想修改要#超级用户可以在忘记密码时使用先创建一个用户addauth digest cxk:cxk666然后修改配置文件/app/zookeeper/bin/zkserver.sh添加超级用户再重启文件./zkserver.sh restart