1. kafka介绍
1.1. 主要功能
根据官网的介绍，ApacheKafka®是一个分布式流媒体平台，它主要有3种功能：
1：It lets you publish and subscribe to streams of records.发布和订阅消息流，这个功能类似于消息队列，这也是kafka归类为消息队列框架的原因
2：It lets you store streams of records in a fault-tolerant way.以容错的方式记录消息流，kafka以文件的方式来存储消息流
3：It lets you process streams of records as they occur.可以再消息发布的时候进行处理
1.2. 使用场景
1：Building real-time streaming data pipelines that reliably get data between systems or applications.在系统或应用程序之间构建可靠的用于传输实时数据的管道，消息队列功能
2：Building real-time streaming applications that transform or react to the streams of data。构建实时的流数据处理程序来变换或处理数据流，数据处理功能
1.3. 详细介绍
Kafka目前主要作为一个分布式的发布订阅式的消息系统使用，下面简单介绍一下kafka的基本机制
1.3.1 消息传输流程
Producer即生产者，向Kafka集群发送消息，在发送消息之前，会对消息进行分类，即Topic，上图展示了两个producer发送了分类为topic1的消息，另外一个发送了topic2的消息。
Topic即主题，通过对消息指定主题可以将消息分类，消费者可以只关注自己需要的Topic中的消息
Consumer即消费者，消费者通过与kafka集群建立长连接的方式，不断地从集群中拉取消息，然后可以对这些消息进行处理。
从上图中就可以看出同一个Topic下的消费者和生产者的数量并不是对应的。
1.3.2 kafka服务器消息存储策略
谈到kafka的存储，就不得不提到分区，即partitions，创建一个topic时，同时可以指定分区数目，分区数越多，其吞吐量也越大，但是需要的资源也越多，同时也会导致更高的不可用性，kafka在接收到生产者发送的消息之后，会根据均衡策略将消息存储到不同的分区中。
在每个分区中，消息以顺序存储，最晚接收的的消息会最后被消费。
1.3.3 与生产者的交互
生产者在向kafka集群发送消息的时候，可以通过指定分区来发送到指定的分区中
也可以通过指定均衡策略来将消息发送到不同的分区中
如果不指定，就会采用默认的随机均衡策略，将消息随机的存储到不同的分区中
1.3.4 与消费者的交互
在消费者消费消息时，kafka使用offset来记录当前消费的位置
在kafka的设计中，可以有多个不同的group来同时消费同一个topic下的消息，如图，我们有两个不同的group同时消费，他们的的消费的记录位置offset各不项目，不互相干扰。
对于一个group而言，消费者的数量不应该多余分区的数量，因为在一个group中，每个分区至多只能绑定到一个消费者上，即一个消费者可以消费多个分区，一个分区只能给一个消费者消费
因此，若一个group中的消费者数量大于分区数量的话，多余的消费者将不会收到任何消息
2、kafka 安装
2.1 jdk安装
#以oracle jdk为例，下载地址http://java.sun.com/javase/downloads/index.jsp
| 1 | yum -y install jdk-8u141-linux-x64.rpm |
2.2 安装zookeeper
| 123 | wget http://apache.forsale.plus/zookeeper/zookeeper-3.4.9/zookeeper-3.4.9.tar.gztar zxf zookeeper-3.4.9.tar.gzmv zookeeper-3.4.9 /data/zk |
修改配置文件内容如下所示:
| 1234567891011121314 | [root@localhost ~]# cat /data/zk/conf/zoo.cfgtickTime=2000initLimit=10syncLimit=5dataDir=/data/zk/data/zookeeperdataLogDir=/data/zk/data/logsclientPort=2181maxClientCnxns=60autopurge.snapRetainCount=3autopurge.purgeInterval=1 server.1=zk01:2888:3888server.2=zk02:2888:3888server.3=zk03:2888:3888 |
参数说明：
server.id=host:port:port:表示了不同的zookeeper服务器的自身标识，作为集群的一部分，每一台服务器应该知道其他服务器的信息。用户可以从“server.id=host:port:port” 中读取到相关信息。在服务器的data(dataDir参数所指定的目录)下创建一个文件名为myid的文件，这个
文件的内容只有一行，指定的是自身的id值。比如，服务器“1”应该在myid文件中写入“1”。这个id必须在集群环境中服务器标识中是唯一的，且大小在1～255之间。这一样配置中，zoo1代表第一台服务器的IP地址。第一个端口号（port）是从follower连接到leader机器的
端口，第二个端口是用来进行leader选举时所用的端口。所以，在集群配置过程中有三个非常重要的端口：clientPort：2181、port:2888、port:3888。
关于zoo.cfg配置文件说明，参考连接https://zookeeper.apache.org/doc/r3.4.10/zookeeperAdmin.html#sc_configuration；
如果想更换日志输出位置，除了在zoo.cfg加入”dataLogDir=/data/zk/data/logs”外，还需要修改zkServer.sh文件，大概修改方式地方在125行左右，内容如下：
| 1234 | 125 ZOO_LOG_DIR="$($GREP "^[[:space:]]*dataLogDir" "$ZOOCFG" | sed -e 's/.*=//')"126 if [ ! -w "$ZOO_LOG_DIR" ] ; then127 mkdir -p "$ZOO_LOG_DIR"128 fi |
在启动服务之前，还需要分别在zookeeper创建myid，方式如下：
| 1 | echo 1 > /data/zk/data/zookeeper/myid |
启动服务
| 1 | /data/zk/bin/zkServer.sh start |
验证服务
| 1234 | ### 查看相关端口号[root@localhost ~]# ss -lnpt|grep javaLISTEN 0 50 :::34442 :::* users:(("java",pid=2984,fd=18))LISTEN 0 50 ::ffff:192.168.15.133:3888 :::* users:(("java",pid=2984,fd=26))LISTEN 0 50 :::2181 :::* users:(("java",pid=2984,fd=25))###查看zookeeper服务状态 |
[root@localhost ~]# /data/zk/bin/zkServer.sh status
ZooKeeper JMX enabled by default
| 1 | Using config: /data/zk/bin/../conf/zoo.cfgMode: follower |
zookeeper相关命令说明，参考https://zookeeper.apache.org/doc/r3.4.10/zookeeperStarted.html （文末有说明）；
2.3 安装kafka
| 12 | tar zxf kafka_2.11-0.11.0.0.tgzmv kafka_2.11-0.11.0.0 /data/kafka |
修改配置
| 123456789101112131415161718192021222324 | [root@localhost ~]# grep -Ev "^#|^$" /data/kafka/config/server.propertiesbroker.id=0delete.topic.enable=truelisteners=PLAINTEXT://192.168.15.131:9092num.network.threads=3num.io.threads=8socket.send.buffer.bytes=102400socket.receive.buffer.bytes=102400socket.request.max.bytes=104857600log.dirs=/data/kafka/datanum.partitions=1num.recovery.threads.per.data.dir=1offsets.topic.replication.factor=1transaction.state.log.replication.factor=1transaction.state.log.min.isr=1log.flush.interval.messages=10000log.flush.interval.ms=1000log.retention.hours=168log.retention.bytes=1073741824log.segment.bytes=1073741824log.retention.check.interval.ms=300000zookeeper.connect=192.168.15.131:2181,192.168.15.132:2181,192.168.15.133:2181zookeeper.connection.timeout.ms=6000group.initial.rebalance.delay.ms=0 |
提示：其他主机将该机器的kafka目录拷贝即可，然后需要修改broker.id、listeners地址。有关kafka配置文件参数，参考：http://orchome.com/12；
启动服务
| 1 | /data/kafka/bin/kafka-server-start.sh /data/kafka/config/server.properties |
验证服务
| 12345 | ### 随便在其中一台主机执行/data/kafka/bin/kafka-topics.sh --create --zookeeper 192.168.15.131:2181,192.168.15.132:2181,192.168.15.133:2181 --replication-factor 1 --partitions 1 --topic test ###在其他主机查看/data/kafka/bin/kafka-topics.sh --list --zookeeper 192.168.15.131:2181,192.168.15.132:2181,192.168.15.133:2181 |
转载自 https://www.cnblogs.com/hei12138/p/7805475.html，https://www.cnblogs.com/yangxiaoyi/p/7359236.html
