基于Flink流处理的动态实时(二) 等有时间了继续写,正好今天有空就写一下服务器配置zookpeer和kafka (单机的)

我用的是：zookeeper-3.4.10.tar.gz 和kafka_2.11-2.1.0.tgz的压缩包,压缩包可以去官方找,也可以

kafka_2.11-2.1.0.tgz的压缩包 https://download.csdn.net/download/qq_37716298/10833302

zookeeper-3.4.10.tar.gz的压缩包 Index of /apache/zookeeper/zookeeper-3.4.10

下载之后可以在服务器上新建一个文件夹,用于存放压缩包。

mkdri test(新建一个test文件夹)

首先安装zookeeper

1.解压 tar -zxvf zookeeper

配置单机zookeeper

2.mv zookeeper /usr/local/zk

3.cd /usr/local/zk/conf

4.cp zoo_sample.cfg zoo.cfg

5.vim zoo.cfg

6.将datadir =/usr/local/zk/data (wq修改保存)

7.在 zk 下面创建data

8.配置环境变量 vim /etc/profile

在java_home中添加zk_home =/usr/local/zk

在path中添加zk的bin目录

9.source /etc/profile

10.进入bin目录 启动zk zkServer.sh start

然后查询当前节点状态 出现standalone 就正确了

进入bin目录，启动、停止、重启分和查看当前节点状态（包括集群中是何角色）别执行：./zkServer.sh start,./zkServer.sh stop,./zkServer.sh restart,./zkServer.sh status

然后安装kafka (回到新建的文件夹,放压缩包那个文件夹)

1.解压 tar -zxvf kafka

2.mv kafka /usr/local

3.cd /usr/local/kafka/config

4.vim server.properties

5.将log.dirs = /usr/local/kafka_2.11-2.1.0/data/kafka-logs

将connect中的localhost修改为自己的ip地址

advertised.listeners=PLAINTEXT://http://xx.xx.xx.xxx:9092(xx代表你服务器的ip)

6.常规模式启动kafka ./kafka-server-start.sh ../config/server.properties &(进入kafka bin目录,(下面也一样))

关闭 kafka ./kafka-server-stop.sh

进程守护模式启动kafka nohup ./kafka-server-start.sh config/server.properties >/dev/null 2>&1 &

如果出现错误：

          Java HotSpot(TM) 64-Bit Server VM warning: INFO: os::commit_memory(0x00000000c0000000, 1073741824, 0) failed; error='Cannot allocate memory' (errno=12)
#
# There is insufficient memory for the Java Runtime Environment to continue.
# Native memory allocation (mmap) failed to map 1073741824 bytes for committing reserved memory.
# An error report file with more information is saved as:
# /usr/local/kafka-2.1.0-src/bin/hs_err_pid11095.log

解决方法

打开/opt/kafka_2.11-2.0.0/binkafka-server-start.sh，降低启动内存大小.

如：export KAFKA_HEAP_OPTS=”-Xmx512M -Xms512M”

7.在启一个窗口 查看是否好用 进入kafka的bin目录

8. ./kafka-topics.sh –create –zookeeper 127.0.0.1:2181 –partitions 1 –replication-factor 1 –topic test(创建一个kafka脚本 一个分区 一个负载因子 主题名字 )

9. ./kafka-console-producer.sh –broker-list 127.0.0.1:9092 –topic test(生产者)

10.在启一个窗口用于消费者：./kafka-console-consumer.sh –broker-list 127.0.0.1:9092 –topic test

如果出现错误

               broker-list is not a recognized option。。。。。

用这个：./kafka-console-consumer.sh –bootstrap-server 127.0.0.1:9092 –topic test

11. centos7中关闭防火墙:systemctl stop firewalld.service

12.说一个有点比较重要的 ：也是花了我半天的功夫 可能是我没这么玩服务器的原因吧 一定要记得把9092端口设置在安全组中 一定要！！！！！！