互联网公司必备的消息中间件、Kafka安装及快速入门

今天带各位老铁对kafka入个门，kafka的集群搭建下，也不知道多少老铁使用过kafka。其实用过的老铁应该没多少。我相信大多老铁用过activeMq，rabbitMq或者rocketMq，这些都是java开发的比较传统的，而且用起来非常简单，结构没那么复杂。很多人都是写业务代码没接触过大数据量高并发的。之前说过rocketMq的历史，它的前身就是metaQ，metaQ来自哪里知道不老铁，其实就是借鉴了kafka，基本上metaQ的第一版就是超的kafka。2010年底kafka开源后，阿里立刻行动通过java模仿了一版kafka名字叫metaQ。作者的名字killme2008，目前在蚂蚁金服。

（一）kafka介绍

• 官网

http://kafka.apache.org/

• 历史

kafka历史

1. 2010年底，开源到github，初始版本为0.7.0。
2. 2011年7月因为备受关注，被纳入apache孵化器项目。
3. 2012年10月，kafka从apache孵化器项目毕业，成为apache顶级项目。
4. 2014年，jay kreps，neha narkhede，jun rao离开linkedin，成立confluent，此后linkedin和confluent成为kafka的核心贡组织，致力于将kafka推广应用。
5. 目前最新版本：2.3.0

• 介绍

Kafka是一种高吞吐量的分布式发布订阅消息系统，使用Scala编写。 scala也属于jvm系统的语言。scala需要寄居在java上运行的。

1. 对于熟悉JMS（Java Message Service）规范的来说，消息系统已经不是什么新概念了(例如ActiveMQ是遵从JMS的规范，RabbitMQ是遵从AMQP的规范等)。
2. Kafka拥有作为一个消息系统应该具备的功能，但是确有着独特的设计。可以这样来说，Kafka借鉴了JMS规范的思想，但是确并没有完全遵循JMS规范。

• 基础术语

1. Topic: Kafka按照Topic分类来维护消息。这个跟JMS的订阅者有些不同，理解为收发消息的主键。
2. Producer： 我们将发布(publish)消息到Topic的进程称之为生产者(producer)。消息的发送者。
3. Consumer：我们将订阅(subscribe)Topic并且处理Topic中消息的进程称之为消费者(consumer)。消费的消费者。
4. Broker： Kafka以集群的方式运行，集群中的每一台服务器称之为一个代理(broker)。activeMQ有服务端，这个就类似kafka的一个节点的服务端。如果多个Broker，就可以组成一个kafka的集群。

• Topic深入理解

Topic是一个类别的名称，所有的message发送到Topic下面。对于每一个Topic，kafka集群按照如下方式维护一个分区(Partition,可以将消息就理解为一个队列Queue)日志文件。

1. 生产者发送消息。其实就是发送给Topic中的某个Partition分区，消费者消费Topic中的某个或者多个Partition分区的消息。某个Partition分区存储的就是咱们实打实的消息。每个Partition可以理解为一个简单的消息队列，生产者发送消息其实是往Partition的末尾发送消息的，最新的消息是发送尾部的，并且每个消息在每个Partition中都有一个唯一的标识符。消费者总是不从头部进行监听的，从old里面取监听消息。
2. 传统的消息中间件，都是消费完直接就不存在了，其实kafka的消费方式不同，kafka有个消费偏移offset的概念，kafka是从偏移量开始往队列的尾部进行消费，在启动消费者如果上图Partition0的偏移量是3，就从3开始消费。每次消费完offset就+1，标识你这个消费者在某个分区消费的位置，下次消费是根据offset记录的位置继续去消费，不会重复消费。
3. 注意：offset是跟消费者有关系的，也就是A消费者offset可能是5，B消费者的offset是从0开始的。也就是说分区是无状态的他跟消费者和生产者是松耦合的关系。也就是生产者只管放就可以了，让消费端来维护offset，这样才可以更好的满足高并发。
4. Topic 其实是个抽象的东西，其实有用的还是里面的分区，分区实实在在存储消息的，一个分区可以理解为一个队列Queue。分区里面的消息不会一直存在的，kafka有个处理过期消息的时间设置，默认是2天时间，根据自我的消费时间，来设置过期时间，合理化的安排防止消息丢失，也可以增加kafka性能。

（二）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

• 修改host文件

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的基本原理。