一 、 Zookeeper 概述

  ZooKeeper是一个分布式的，开放源码的分布式应用程序协调服务，是Google的Chubby一个开源的实现，是Hadoop和Hbase的重要组件。它是一个为分布式应用提供一致性服务的软件，提供的功能包括：配置维护、域名服务、分布式同步、组服务等。

  ZooKeeper的目标就是封装好复杂易出错的关键服务，将简单易用的接口和性能高效、功能稳定的系统提供给用户。

  Zookeeper的官方网站可以学习

1. 下载Zookeeper

   wget https://mirrors.tuna.tsinghua.edu.cn/apache/zookeeper/zookeeper-3.5.6/apache-zookeeper-3.5.6-bin.tar.gz

2. 安装Zookeeper

~$ mkdir /usr/local/zookeeper
~$ cd /usr/local/zookeeper
~$ tar -zxvf ~/apache-zookeeper-3.5.6-bin.tar.gz

3. 配置Zookeeper

~4 cd apache-zookeeper-3.5.6-bin
~$ cp conf/zoo_sample.cfg conf/zoo.cfg
~$ vi conf/zoo.cfg
admin.serverPort=8888

4. 设置开机启动

   ~$ vi /etc/profile export ZOOKEEPER_HOME=/usr/local/zookeeper/apache-zookeeper-3.5.6-bin/ export PATH=$ZOOKEEPER_HOME/bin:$PATH ~$ source /etc/profile ~$ cd /etc/rc.d/init.d ~$ touch zookeeper ~$ vi /etc/rc.d/init.d/zookeeper #!/bin/bash #chkconfig: 2345 10 90 #description: service zookeeper export JAVA_HOME=/usr/java/jdk1.8.0_172 ZOOKEEPER_HOME=/usr/local/zookeeper/apache-zookeeper-3.5.6/ export ZOO_LOG_DIR=/usr/local/zookeeper/log su root ${ZOOKEEPER_HOME}/bin/zkServer.sh "$1" ~$ chmod +x /etc/rc.d/init.d/zookeeper ~$ chkconfig --add zookeeper ~$ service zookeeper start

5. 检查测试Zookeeper

   # 查看状态 ~$ service zookeeper status # 查看2181端口是否启用 ~$ lsof -i:2181 # 查看网络状态 ~$ netstat -lntup

二、Kafka 概述

  Kafka是由Apache软件基金会开发的一个开源流处理平台，由Scala和Java编写。Kafka是一种高吞吐量的分布式发布订阅消息系统，它可以处理消费者规模的网站中的所有动作流数据。 这种动作（网页浏览，搜索和其他用户的行动）是在现代网络上的许多社会功能的一个关键因素。 这些数据通常是由于吞吐量的要求而通过处理日志和日志聚合来解决。 对于像Hadoop的一样的日志数据和离线分析系统，但又要求实时处理的限制，这是一个可行的解决方案。Kafka的目的是通过Hadoop的并行加载机制来统一线上和离线的消息处理，也是为了通过集群来提供实时的消息.Kafka 支持Java及多种其它语言客户端，可与Hadoop、Storm、Spark等其它大数据工具结合使用。

• 下载Kafka

  wget http://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.3.1/kafka_2.12-2.3.1.tgz

• 安装Kafka

  ~$ mkdir /usr/local/kafka ~$ tar -zxvf ~/kafka_2.12-2.3.1.tgz

• 配置Kafka

  ~$ cd /usr/local/server/kafka_2.12-2.3.1/config/ ~$ vi server.properties

• 验证

  • 后台启动zookeeper

    ~$ /usr/local/kafka/kafka_2.12-2.3.1 ~$ bin/zookeeper-server-start.sh -daemon config/zookeeper.propertie

  • 后台启动kafka

    ~$ /usr/local/kafka/kafka_2.12-2.3.1 ~$ bin/kafka-server-start.sh -daemon config/server.properties

  • 创建topic

    使用kafka-topics.sh 创建单分区单副本的topic test

    ~$ bin/kafka-topics.sh --create --zookeeper 192.168.199.254:2181 --replication-factor 1 --partitions 1 --topic test

  • 产生消息命令

    使用kafka-console-producer.sh 命令向topic test发送消息

    ~$ bin/kafka-console-producer.sh --broker-list 192.168.199.254:9092 --topic test

  • 消费消息

    使用kafka-console-consumer.sh 命令接受来自topic test的消息

    ~$ bin/kafka-console-consumer.sh --bootstrap-server 192.168.199.254:9092 --topic test --from-beginning

• 设置开机启动Kafka

  • 在/etc/rc.d/init.d新建一个kafka文件，内容如下

    ~$ cd /etc/rc.d/init.d ~$ touch kafka ~$ vi kafka #!/bin/sh # # chkconfig: 2345 10 90 # description: Kafka # # File : Kafka # # Description: Starts and stops the Kafka server for SASL/PLAIN # source /etc/rc.d/init.d/functions SERVER_HOME=/usr/local/server KAFKA_HOME=${SERVER_HOME}/kafka_2.12-2.3.1/bin LOG_PATH=${SERVER_HOME}/log KAFKA_USER=root export JRE_HOME=/usr/java/jdk1.8.0_172/jre export JAVA_HOME=/usr/java/jdk1.8.0_172 export CLASSPATH=$JAVA_HOME/lib/tools.jar:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib export PATH=$JAVA_HOME/bin:$PATH [ -e /etc/sysconfig/kafka ] && . /etc/sysconfig/kafka # See how we were called. case "$1" in start) echo -n "Starting Kafka:" /sbin/runuser -s /bin/sh $KAFKA_USER -c "nohup $KAFKA_HOME/bin/kafka-server-start.sh $KAFKA_HOME/config/server.properties > $LOG_PATH/server.out 2> $LOG_PATH/server.err &" echo " done." exit 0 ;; stop) echo -n "Stopping Kafka: " /sbin/runuser -s /bin/sh $KAFKA_USER -c "ps -ef | grep kafka.Kafka | grep -v grep | awk '{print \$2}' | xargs kill" echo " done." exit 0 ;; hardstop) echo -n "Stopping (hard) Kafka: " /sbin/runuser -s /bin/sh $KAFKA_USER -c "ps -ef | grep kafka.Kafka | grep -v grep | awk '{print \$2}' | xargs kill -9" echo " done." exit 0 ;; status) c_pid=`ps -ef | grep kafka.Kafka | grep -v grep | awk '{print $2}'` if [ "$c_pid" = "" ] ; then echo "Stopped" exit 3 else echo "Running $c_pid" exit 0 fi ;; restart) stop start ;; *) echo "Usage: kafka {start|stop|hardstop|status|restart}" exit 1 ;; esac

    • 赋予脚本可执行权限

      chmod +x kafka

  • 添加到开机启动项

    chkconfig --add kafka

  • 设置开机启动

    chkconfig kafka on

  • 查看开机启动项，确认kafka已添加

    chkconfig --list

• 监控插件 kafka eagle

1. 安装包下载

   ~$ wget https://github.com/smartloli/kafka-eagle-bin/archive/v1.4.1.tar.gz ~$ cd /usr/local/kafka ~$ tar -zxvf ~/v1.4.1.tar.gz ~$ mv v1.4.1 afka-eagle-1.4.1

2. 修改环境变量，增加 JAVA_HONE 和 KE_HOME

   ~$ vi /etc/profile #kafka-eagle-web-1.3.0 export KE_HOME=/usr/local/kafka/kafka-eagle-web-1.4.1/ export PATH=$PATH:$KE_HOME/bin ~$ source /etc/profile

3. 修改 kafka-eagle 配置文件

   vi conf/system-config.properties kafka.eagle.zk.cluster.alias=cluster1 cluster1.zk.list=192.168.16.139:2181 kafka.eagle.webui.port=8048 kafka.eagle.metrics.charts=true kafka.eagle.sql.fix.error=true kafka.eagle.driver=com.mysql.jdbc.Driver kafka.eagle.url=jdbc:mysql://127.0.0.1:3306/kafka?useSSL=true&characterEncoding=utf8&useUnicode=true&serverTimezone=UTC kafka.eagle.username=username kafka.eagle.password=password

4. 赋予脚本可执行权限

   ~$ chmod +x bin/ke.sh ~$ ./bin/ke.sh start

5. 设置开机启动kafka-eagle

   ~$ vi /etc/init.d/kafka-eagle #!/bin/bash #chkconfig:2345 80 20 #description:kafka-eagle KAFKA_EAGLE_HOME=/usr/local/kafka/kafka-eagle-web-1.4.1/ case $1 in         start) ${KAFKA_EAGLE_HOME}/bin/ke.sh start ;;         stop) ${KAFKA_EAGLE_HOME}/bin/ke.sh stop ;;         status) ${KAFKA_EAGLE_HOME}/bin/ke.sh status ;;         restart) ${KAFKA_EAGLE_HOME}/bin/ke.sh restart ;;         *) echo "require start|stop|status|restart" ;; esac ~$ chmod +x kafka-eagle ~$ chkconfig --add /etc/init.d/kafka-eagle

• 安装配置kafka-manager集群管理工具`

  kafka-manager是最受欢迎的kafka集群管理工具，事由雅虎开源的一个项目。用户可以在web管理界面执行一些简单的集群管理操作。具体支持以下内容：

• 管理多个集群
• 轻松管理集群状态（主题、消费者、偏移、代理、副本分发、分区分发）
• 运行首选放油阀选举
• 使用选项生成分区分配以选择使用的的代理
• 运行分区重新分配（基于生产的分配）
• 使用可选主题配置创建主题
• 删除主题
• 主题列表现在指示标记为删除主题
• 批量生成多个主题的分区分配，并可选择要使用的代理
• 批量运行重新分配多个主题的分区
• 将分区添加到现有主题
• 更新现有主题的配置

  github.com项目地址：kafka-manager项目