Kafka教程

简介

1.消息中间件介绍

        消息中间件是在消息的传输过程中保存消息的容器。消息中间件在将消息从消息生产者到消费者时充当中间人的作用。队列的主要目的是提供路由并保证消息的传送；如果发送消息时接收者不可用，消息对列会保留消息，直到可以成功地传递它为止，当然，消息队列保存消息也是有期限的

2.消息中间件特点

1. 采用异步处理模式：
   消息发送者可以发送一个消息而无须等待响应。消息发送者将消息发送到一条虚拟的通道（主题或者队列）上，消息接收者则订阅或者监听该通道。一条消息可能最终转发给一个或多个消息接收者，这些接收者都无需对消息发送者做出同步回应。整个过程是异步的。 比如用户信息注册。注册完成后过段时间发送邮件或者短信。
2. 应用程序和应用程序调用关系为松耦合关系：
   发送者和接收者不必要了解对方、只需要确认消息 发送者和接收者不必同时在线 比如在线交易系统为了保证数据的最终一致，在支付系统处理完成后会把支付结果放到信息中间件里通知订单系统修改订单支付状态。两个系统通过消息中间件解耦。

3.Kafka介绍

        Apache kafka是消息中间件的一种，举个例子，生产者消费者，生产者生产鸡蛋，消费者消费鸡蛋，生产者生产一个鸡蛋，消费者就消费一个鸡蛋，假设消费者消费鸡蛋的时候噎住了（系统宕机了），生产者还在生产鸡蛋，那新生产的鸡蛋就丢失了。再比如生产者很强劲（大交易量的情况），生产者1秒钟生产100个鸡蛋，消费者1秒钟只能吃50个鸡蛋，那要不了一会，消费者就吃不消了（消息堵塞，最终导致系统超时），消费者拒绝再吃了，”鸡蛋“又丢失了，这个时候我们放个篮子在它们中间，生产出来的鸡蛋都放到篮子里，消费者去篮子里拿鸡蛋，这样鸡蛋就不会丢失了，都在篮子里，而这个篮子就是”kafka“。 鸡蛋其实就是“数据流”，系统之间的交互都是通过“数据流”来传输的（就是tcp、http什么的），也称为报文，也叫“消息”。 消息队列满了，其实就是篮子满了，”鸡蛋“ 放不下了，那赶紧多放几个篮子，其实就是kafka的扩容。 现在知道kafka是干什么的了吧，它就是那个"篮子"。

4.kafka名词解释

• producer：生产者，就是它来生产“鸡蛋”的。
• consumer：消费者，生出的“鸡蛋”它来消费。
• topic：你把它理解为标签，生产者每生产出来一个鸡蛋就贴上一个标签（topic），消费者可不是谁生产的“鸡蛋”都吃的，这样不同的生产者生产出来的“鸡蛋”，消费者就可以选择性的“吃”了。
• broker：就是篮子了。 
  
  一定要学会抽象的去思考，上面只是属于业务的角度，如果从技术角度，topic标签实际就是队列，生产者把所有“鸡蛋（消息）”都放到对应的队列里了，消费者到指定的队列里取。

入门配置

1.Ubuntu下安装和使用

• 下载地址：http://kafka.apache.org/downloads

 

• 把tgz包放到opt目录下面

• 解压安装包到/opt/kafka目录下

• 进入到/opt/kafka/kafka_2.11-2.0.0目录下启动kafka自带的zookeeper

  /bin/zookeeper-server-start.sh config/zookeeper.properties

 

• 启动zookeeper成功，2181端口

 

• 打开另一个终端查看2181端口已经被监听

 

• 箭头就是zookeeper开启的服务

 

• 用下面命令启动kafka,9092端口

  bin/kafka-server-start.sh config/server.properties

• 再打开另一个终端查看9092端口已经被监听,zookeeper端口也把kafka服务加载进去

 

• 输入jps命令发现服务都开启了

• 创建一个主题，名字叫test1

  bin/kafka-topics.sh --zookeeper localhost:2181 --create --topic test1 --partitions 3 --replication-factor 1

 

• 查看test1主题被创建成功,创建了三个Partition

 

• 启动生产者（producer），在生产者中发送的消息，会存储在broker中，需要消费者接收这些消息。

  bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test1

 

• 启动消费者（consumer），消费者接收到，刚刚生产者输出的信息

  bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test1

 

 

• 查看生产者的历史信息

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test1 --from-beginning

 

• kafka的默认端口是9092，在config/server.properties也可以修改这里去掉#自己定义

• zookeeper的默认端口是2181，在config/zookeeper.propertie也可以修改这里自己定义

 

不用kafka自带的zookeeper,下载一个zookeeper来启动

 

• 解压到opt下

 

• 启动zookeeper

  bin/zkServer.sh start conf/zoo_sample.cfg

 

• 查看状态

  bin/zkServer.sh status conf/zoo_sample.cfg

• 启动成功

• 然后像上面一样启动kafka就行了