[大数据]kafka-工具脚本 01

/usr/bin下面的是系统用户使用的应用程序

指定参数

上面的这些kafka运维脚本，有些是指定参数–zookeeper，有些是指定参数–broker-list，有些是指定参数–bootstrap-server。

这实际上是历史问题。–broker-list代表broker地址，而–bootstrap-server代表连接起点，可以从中拉取broker地址信息。bootstrap-server的命名更高级点。还有通过zookeeper连接的，kafka早起很多信息存方在zk中，后期慢慢弱化了zk的作用，这三个参数代表kafka的三个时代。

1、kafka节点的启/停工具

kafka-server-start.sh 启动节点，使用方法：nohup ./bin/kafka-server-start.sh ./config/server.properties &

kafka-server-start首先判断是否有配置文件信息，然后找到log4j配置，其次分配java进程的head大小，最后启动kafka-run-class.sh kafka.kafka脚本，kafka.kafka.scala位于core目录下。

// 首先判断命令后面是否有server.properties,没有的话，就会直接的退出
# 检查参数个数
if [ $# -lt 1 ];
then
	echo "USAGE: $0 [-daemon] server.properties [--override property=value]*"
	exit 1
fi
//获取当前文件夹
base_dir=$(dirname $0)

// 判断KAFKA_LOG4J_OPTS为空的话，就将kafka自带的配置文件赋值给日志文件变量
// 配置文件在当前目录的上一层目录的下面的config里面的log4j.properties
# 设置Log4j相关的环境变量
if [ "x$KAFKA_LOG4J_OPTS" = "x" ]; then
    export KAFKA_LOG4J_OPTS="-Dlog4j.configuration=file:$base_dir/../config/log4j.proper
fi

// 如果KAFKA_HEAP_OPTS变量为空的话，就配置默认值，默认head的xms初始是1G，xmx最大为1G，设置相等是为了防止内存的抖动。
// "x$KAFKA_HEAP_OPTS" = "x" 其实就是判断$KAFKA_HEAP_OPTS是否为空的,加上x是因为如果直接
// 判断"$KAFKA_HEAP_OPTS" = "" 某些版本的bash可能会出现编译错误，这里是为了兼容不同的版本的。
# 设置JVM的内存
if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
    export KAFKA_HEAP_OPTS="-Xmx1G -Xms1G"
fi

EXTRA_ARGS=${EXTRA_ARGS-'-name kafkaServer -loggc'}
# 检查第一个参数是否为“daemon”
COMMAND=$1
case $COMMAND in
  -daemon)
    EXTRA_ARGS="-daemon "$EXTRA_ARGS
    shift //
    ;;
  *)
    ;;
esac

//执行当前目录项目的kafka-run-class.sh脚本来运行kafka.Kafka类
//这里的$@和$*的含义是一样的，都是表示可以脚本后面的所有的参数
exec $base_dir/kafka-run-class.sh $EXTRA_ARGS kafka.Kafka "$@"
# 很多脚本和工具类都依赖于kafka-run-class.sh脚本，其主要的功能是设置CLASSPATH，进行JMX的相关配置，配置Log4j，
指定存放日志文件和索引文件位置，检测JAVA HOME环境变量，进行JVM的相关配置，决定是否后台启动。

补充

JMX(Java Management Extensions)Java 管理扩展，用来管理和监测 Java 程序。

JMX 既是 Java 管理系统的一个标准，一个规范，也是一个接口，一个框架。有标准、有规范是为了让开发者可以定制开发自己的扩展功能，而且作为一个框架来讲，JDK 已经帮我们实现了常用的功能，尤其是对 JVM 的监控和管理。

MBean

JMX 是通过各种 MBean(Managed Bean) 传递消息的，MBean 其实就是我们经常说的 Java Bean，只不过由于它比较特殊，所以称之为 MBean。既然是个 Bean，里面就是一些属性和方法，外界就可以获取被管理的资源的状态和操纵MBean的行为。JMX 中共有四种类型的 MBean，分别是 Standard MBean, Dynamic MBean, Open MBean, Model MBean。JDK 提供的 MBean 主要在 java.lang.management 和 javax.management包里面。可以进去看一看，进去就能看到好多似曾相识的身影，比如 Memory 相关的、Thread 相关的，这不就是我们在 VisualVM 上看到的内容吗，没错，数据就是从这里来的。

kafka-server-stop.sh 节点停止

kafka-server-stop.sh会找到所有kafka进程杀掉，所以如果一个机器上有多个kafka进程，可以找到进程号，执行kill –s TERM $pid，完成进程关闭；

bin/kafka-server-stop.sh
PIDS=$(ps ax | grep -i 'kafka\.Kafka' | grep java | grep -v grep | awk '{print $1}')

2、kafka-connect 数据传输工具

kafka connect 是一个可扩展的、可靠的在kafka和其他系统之间流传输的数据工具。简而言之就是他可以通过Connector（连接器）简单、快速的将大集合数据导入和导出kafka。可以接收整个数据库或收集来自所有的应用程序的消息到kafka的topic中
参考：https://www.cnblogs.com/tree1123/p/11434047.html
https://www.cnblogs.com/boanxin/p/10088898.html

connect-mirror-maker.sh 脚本
mirror maker2背景：通常情况下，我们都是使用一套kafka集群处理业务。但有些情况需要使用另一套kafka集群来进行数据同步和备份。在kafka早先版本的时候，kafka针对这种场景就有推出一个叫mirror maker的工具，用来同步两个kafka集群的数据。

kafka-mirror-maker.sh //用于在Kafka集群间实现数据镜像
介绍：MirrorMaker是Kafka附带的一个用于在Kafka集群之间制作镜像数据的工具。该工具从源集群中消费并生产到目标群集。这种镜像的常见用例是在另一个数据中心提供副本。
其实现原理是通过从源集群中消费消息，然后将消息生产到目标集群中，也就是普通的生产和消费消息。
用户只需要在启动Kafka Mirror Maker时指定一些简单的消费端和生产端配置就可以实现准实时的数据同步。
参考：https://www.cnblogs.com/listenfwind/p/14269259.html
https://blog.csdn.net/qq_41154944/article/details/108282641