今天带各位老铁对kafka入个门,kafka的集群搭建下,也不知道多少老铁使用过kafka。其实用过的老铁应该没多少。我相信大多老铁用过activeMq,rabbitMq或者rocketMq,这些都是java开发的比较传统的,而且用起来非常简单,结构没那么复杂。很多人都是写业务代码没接触过大数据量高并发的。之前说过rocketMq的历史,它的前身就是metaQ,metaQ来自哪里知道不老铁,其实就是借鉴了kafka,基本上metaQ的第一版就是超的kafka。2010年底kafka开源后,阿里立刻行动通过java模仿了一版kafka名字叫metaQ。作者的名字killme2008,目前在蚂蚁金服。
http://kafka.apache.org/
kafka历史
Kafka是一种高吞吐量的分布式发布订阅消息系统,使用Scala编写。 scala也属于jvm系统的语言。scala需要寄居在java上运行的。
Topic是一个类别的名称,所有的message发送到Topic下面。对于每一个Topic,kafka集群按照如下方式维护一个分区(Partition,可以将消息就理解为一个队列Queue)日志文件。
生产者发送消息。其实就是发送给Topic中的某个Partition分区,消费者消费Topic中的某个或者多个Partition分区的消息。某个Partition分区存储的就是咱们实打实的消息。每个Partition可以理解为一个简单的消息队列,生产者发送消息其实是往Partition的末尾发送消息的,最新的消息是发送尾部的,并且每个消息在每个Partition中都有一个唯一的标识符。消费者总是不从头部进行监听的,从old里面取监听消息。
传统的消息中间件,都是消费完直接就不存在了,其实kafka的消费方式不同,kafka有个消费偏移offset的概念,kafka是从偏移量开始往队列的尾部进行消费,在启动消费者如果上图Partition0的偏移量是3,就从3开始消费。每次消费完offset就+1,标识你这个消费者在某个分区消费的位置,下次消费是根据offset记录的位置继续去消费,不会重复消费。
注意:offset是跟消费者有关系的,也就是A消费者offset可能是5,B消费者的offset是从0开始的。也就是说分区是无状态的他跟消费者和生产者是松耦合的关系。也就是生产者只管放就可以了,让消费端来维护offset,这样才可以更好的满足高并发。
Topic 其实是个抽象的东西,其实有用的还是里面的分区,分区实实在在存储消息的,一个分区可以理解为一个队列Queue。分区里面的消息不会一直存在的,kafka有个处理过期消息的时间设置,默认是2天时间,根据自我的消费时间,来设置过期时间,合理化的安排防止消息丢失,也可以增加kafka性能。
由于Kafka是用Scala语言开发的,运行在JVM上,因此在安装Kafka之前需要先安装JDK。
wget --no-cookies --no-check-certificate --header "Cookie: gpw_e24=http%3A%2F%2Fwww.oracle.com%2F; oraclelicense=accept-securebackup-cookie" "http://download.oracle.com/otn-pub/java/jdk/8u141-b15/336fa29ff2bb4ef291e347e091f7f4a7/jdk-8u141-linux-x64.tar.gz" tar -zxvf jdk* cd jdk* echo "export JAVA_HOME=/root/jdk1.8.0_141" >> /etc/profile echo "export PATH=$""JAVA_HOME/bin:$""PATH" >> /etc/profile source /etc/profile
kafka依赖zookeeper,所以需要先安装zookeeper
# 一定要下载带bin的,不要下载不带bin的会爆:Could not find or load main class org.apache.zookeeper.server.quorum.QuorumP cd /opt/ wget http://mirror.bit.edu.cn/apache/zookeeper/stable/apache-zookeeper-3.5.5-bin.tar.gz tar -zxvf apache-zookeeper-3.5.5-bin.tar.gz cd apache-zookeeper-3.5.5-bin cp conf/zoo_sample.cfg conf/zoo.cfg
启动zookeeper
bin/zkServer.sh start conf/zoo.cfg & bin/zkCli.sh ls /
下载1.1.0 release版本
wget http://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.2.1/kafka_2.12-2.2.1.tgz tar -xzf kafka_2.12-2.2.1.tgz cd kafka_2.12-2.2.1
hostname vi /etc/hosts #加入127.0.0.1 kafka
cd /opt/kafka_2.12-2.2.1 bin/kafka-server-start.sh -daemon config/server.properties cd /opt/apache-zookeeper-3.5.5-bin #我们进入zookeeper目录通过zookeeper客户端查看下zookeeper的目录树 bin/zkCli.sh #查看zk的根目录kafka相关节点 ls / #查看kafka节点 ls /brokers/ids
现在我们来创建一个名字为“test”的Topic,这个topic只有一个partition(实际上可以创建多个分区,你改成几就有几个分区),并且备份因子也设置为1,其实就是不做备份,因为就一个不是集群。会发现主题就是一个虚拟的概念,其实还是创建分区。
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
查看kafka中目前存在的topic
cd /opt/kafka_2.12-2.2.1 bin/kafka-topics.sh --list --zookeeper localhost:2181
在zookeeper中也添加了主题目录,看来kafka跟zookeeper简直是形影不离。
kafka自带了一个producer命令客户端,可以从本地文件中读取内容,或者我们也可以以命令行中直接输入内容,并将这些内容以消息的形式发送到kafka集群中。在默认情况下,每一个行会被当做成一个独立的消息。首先我们要运行发布消息的脚本,然后在命令中输入要发送的消息的内容
对于consumer,kafka同样也携带了一个命令行客户端,会将获取到内容在命令中进行输出。如果你是通过不同的终端窗口来运行以上的命令,你将会看到在producer终端输入的内容,很快就会在consumer的终端窗口上显示出来。
# 0.90版本之后启动消费者的方法--bootstrap-server 连接kafka的地址,之前的版本是--zookeeper 连接的是zookeeper的地址 cd /opt/kafka_2.12-2.2.1 bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning
当我们不携带任何参数运行命令的时候,将会显示出这个命令的详细用法。
一个发送,一个接收。
PS:kafka安装完毕了,也介绍了kafka的基本原理。