消息队列介绍

Kafka分布式消息系统被认为是一种消息引擎系统,或者消息队列中间件。

队列(Queque)是一种先入先出(Fist In First Out - FIFO)的线性表数据结构, 可以使用数组或者链表实现队列, 一个队列需要维护两个指针, head指向队首, tail指向队尾, 移动队尾添加元素(入队), 移动队首指针删除元素(出队).

image-20200515093054216

实际生活中,队列的应用随处可见,比如排队、挂号、传递过程都可以用队列来描述或者实现。

什么是消息队列

生产出美味的巧克力需要三道工序:首先将可可豆磨成可可粉,然后将可可粉加热并加入糖变成巧克力浆,最后将巧克力浆灌入模具,撒上坚果碎,冷却后就是成品巧克力了。

最开始的时候,每次研磨出一桶可可粉后,工人就会把这桶可可粉送到加工巧克力浆的工人手上,然后再回来加工下一桶可可粉。这样一来我们很快就发现,其实工人可以不用自己运送半成品,于是在每道工序之间都增加了一组传送带,研磨工人只要把研磨好的可可粉放到传送带上,就可以去加工下一桶可可粉了。 传送带解决了上下游工序之间的“通信”问题。

传送带上线后确实提高了生产效率,但也带来了新的问题:每道工序的生产速度并不相同。在巧克力浆车间,一桶可可粉传送过来时,工人可能正在加工上一批可可粉,没有时间接收。不同工序的工人们必须协调好什么时间往传送带上放置半成品,如果出现上下游工序加工速度不一致的情况,上下游工人之间必须互相等待,确保不会出现传送带上的半成品无人接收的情况。

为了解决这个问题,我们在每组传送的下游带配备了一个暂存半成品的仓库,这样上游工人就不用等待下游工人有空,任何时间都可以把加工完成的半成品丢到传送带上,无法接收的货物被暂存在仓库中,下游工人可以随时来取。传送带配备的仓库实际上起到了“通信”过程中“缓存”的作用。

img

在解决上述问题的过程中, 不知不觉我们就实现了一个消息队列.

队列能解决什么问题?

缓存数据,且保持数据存取顺序一致

为什么使用消息队列

消息异步处理

如需要快速响应的事情集中资源处理,其余的放入消息队列异步完成

很多时候,你不想也不需要立即处理消息。消息队列提供了异步处理机制,允许你把一个消息放入队列,但并不立即处理它。你想向队列中放入多少消息就放多少,然后在你想要处理的时候再去处理它们。

想一想如何设计一个秒杀系统可以达到更高的成交量。

秒杀系统需要解决的核心问题是,如何利用有限的服务器资源,尽可能多地处理短时间内的海量请求。我们知道,处理一个秒杀请求包含了很多步骤,例如:

  • 风险控制;
  • 库存锁定;
  • 生成订单;
  • 短信通知;
  • 更新统计数据。

能否决定秒杀成功,实际上只有风险控制和库存锁定这 2 个步骤。只要用户的秒杀请求通过风险控制,并在服务端完成库存锁定,就可以给用户返回秒杀结果了,对于后续的生成订单、短信通知和更新统计数据等步骤,并不一定要在秒杀请求中处理完成。

所以当服务端完成前面 2 个步骤,确定本次请求的秒杀结果后,就可以马上给用户返回响应,然后把请求的数据放入消息队列中,由消息队列异步地进行后续的操作。

file

处理一个秒杀请求,从 5 个步骤减少为 2 个步骤,这样不仅响应速度更快,并且在秒杀期间,我们可以把大量的服务器资源用来处理秒杀请求。秒杀结束后再把资源用于处理后面的步骤,充分利用有限的服务器资源处理更多的秒杀请求。

可以看到,在这个场景中,消息队列被用于实现服务的异步处理。这样做的好处是:

  • 可以更快地返回结果
  • 减少等待,自然实现了步骤之间的并发,提升系统总体的性能。

流量控制

继续说我们的秒杀系统,我们已经使用消息队列实现了部分工作的异步处理,但我们还面临一个问题:如何避免过多的请求压垮我们的秒杀系统?

加入消息队列后,整个秒杀流程变为:
file
网关在收到请求后,将请求放入请求消息队列;
后端服务从请求消息队列中获取 APP 请求,完成后续秒杀处理过程,然后返回结果。

秒杀开始后,当短时间内大量的秒杀请求到达网关时,不会直接冲击到后端的秒杀服务,而是先堆积在消息队列中,后端服务按照自己的最大处理能力,从消息队列中消费请求进行处理。

这种设计的优点是:能根据下游的处理能力自动调节流量,达到“削峰填谷”的作用。

img

但这样做同样是有代价的:

增加了系统调用链环节,导致总体的响应时延变长。
上下游系统都要将同步调用改为异步消息,增加了系统的复杂度。

服务解耦

消息队列的另外一个作用,就是实现系统应用之间的解耦。

订单是电商系统中比较核心的数据,当一个新订单创建时:

支付系统需要发起支付流程;
风控系统需要审核订单的合法性;
客服系统需要给用户发短信告知用户;
经营分析系统需要更新统计数据;
……

这些订单下游的系统都需要实时获得订单数据。随着业务不断发展,这些订单下游系统不断的增加,不断变化

img

引入消息队列后,订单服务在订单变化时发送一条消息到消息队列的一个主题 Order 中,所有下游系统都订阅主题 Order,这样每个下游系统都可以获得一份实时完整的订单数据。

img

无论增加、减少下游系统或是下游系统需求如何变化,订单服务都无需做任何更改,实现了订单服务与下游服务的解耦。

总结

以上就是消息队列最常被使用的三种场景:异步处理、流量控制和服务解耦。当然,消息队列的适用范围不仅仅局限于这些场景.简单的说,我们在单体应用里面需要用队列解决的问题,在分布式系统中大多都可以用消息队列来解决。

消息队列的选型

  • 开源
  • 流行, 活跃
  • 兼容
  • 确保消息的可靠传递
  • 支持集群, 高可用
  • 性能要求

可供选择的消息队列

1. Rabbit MQ (AMQP协议)
  • "messaging that just works"
  • 老牌产品
2. Rocket MQ
  • Alibaba 2012年开源
  • 2017年成为Apache顶级项目
  • 经过多次双十一考验, 稳定可靠
3. Kafka
  • 也是Apache顶级项目
  • 最初设计目的是处理海量日志
    • 不保证消息可靠性(可能丢消息)
    • 不支持集群
  • 当下的Kafka
    • 非常成熟的消息队列产品
    • 稳定可靠, 功能特性可以满足绝大多数场景要求
    • 兼容性好,尤其是大数据和流计算领域优先支持Kafka!!!
    • 异步性能最好(同步性能反而较差).
4. 第二梯队
  1. Active MQ(JMS)
  2. Zero MQ
  3. 雅虎Pulsar

Views: 240

2-Storm集群安装(伪分布式)

基础环境:百度网盘,提取码:NIIT
--来自百度网盘超级会员V4的分享)

安装ZooKeeper集群

Storm 使用 Zookeeper 协调管理集群. Zookeeper 并不是 用于消息传递, 所以 Storm 对Zookeeper造成的负载压力非常低. 单节点Zookeeper集群在大多数情况下应该是足够的,但是如果您想要故障转移或部署大型Storm集群,则可能需要较大的Zookeeper集群. 部署Zookeeper的说明是这里.

这里我们安装一个单机伪分布式的Zookeeper集群:

conf/zoo_1.cfg

tickTime=2000
initLimit=10
syncLimit=5
dataDir=/home/hadoop/tmp/zk1/data
dataLogDir=/home/hadoop/tmp/zk1/dataLog
clientPort=2181
4lw.commands.whitelist=*
server.1=localhost:2891:3891
server.2=localhost:2892:3892
server.3=localhost:2893:3893

conf/zoo_2.cfg

tickTime=2000
initLimit=10
syncLimit=5
dataDir=/home/hadoop/tmp/zk2/data
dataLogDir=/home/hadoop/tmp/zk2/dataLog
clientPort=2182
4lw.commands.whitelist=*
server.1=localhost:2891:3891
server.2=localhost:2892:3892
server.3=localhost:2893:3893

conf/zoo_3.cfg

tickTime=2000
initLimit=10
syncLimit=5
dataDir=/home/hadoop/tmp/zk3/data
dataLogDir=/home/hadoop/tmp/zk3/dataLog
clientPort=2183
4lw.commands.whitelist=*
server.1=localhost:2891:3891
server.2=localhost:2892:3892
server.3=localhost:2893:3893

在dataDir中创建文件myid内容是1或2或3

echo 1 > /home/hadoop/tmp/zk1/data/myid
echo 2 > /home/hadoop/tmp/zk2/data/myid
echo 3 > /home/hadoop/tmp/zk3/data/myid

启动服务

bin/zkServer.sh start conf/zoo_1.cfg
bin/zkServer.sh start conf/zoo_2.cfg
bin/zkServer.sh start conf/zoo_3.cfg

关于Zookeeper部署的几点注意事项:

1.在监督(supervision)下运行Zookeeper至关重要,因为Zookeeper是快速失败的,如果遇到任何错误的情况都将退出进程. 有关详细信息,请参阅这里 .

2.建立一个cron定时任务来压缩Zookeeper的数据和事务日志至关重要. Zookeeper守护进程本身不会这样做,如果没有设置cron,Zookeeper将很快耗尽磁盘空间. 有关详细信息,请参阅这里.

Nimbus和worker节点的安装环境

接下来你需要准备Nimbus 和 worker 节点的安装环境:

  1. Java 8+ (Apache Storm 2.x is tested through travisci against a java 8 JDK)
  2. Python 2.6.6 (Python 3.x should work too,but is not tested as part of our CI enviornment)

这些依赖版本是Storm已经测试过的. Storm 在不同的Java 或Python版本上也许会存在问题.

下载解压 Storm

接下来,下载一个Storm版本,并解压zip文件到Nimbus和每个worker机器上的某个目录下. Storm版本可以从这里下载.

当前最新的版本是2.2.0,这里使用2.1.0的版本。

我们需要下载:

也可以使用wget下载,如

wget https://dlcdn.apache.org/storm/apache-storm-2.1.0/apache-storm-2.1.0.tar.gz

在storm.yaml中设置必要的配置

Storm 发布包中在目录conf/storm.yaml 下包含一个默认的配置文件. 你可以在[这里](http://github.com/apache/storm/blob/master /conf/defaults.yaml)查看默认值. storm.yaml 中的存在的配置项会覆盖掉 defaults.yaml中相应的配置项. 下面一些配置是集群运行时所必要的:

  1. storm.zookeeper.servers: 这是一个Storm集群所依赖 Zookeeper 集群的hosts列表. 类似于:
storm.zookeeper.servers:
  - "111.222.333.444"
  - "555.666.777.888"

如果配置的Zookeeper集群不是默认的端口, 你应该设置 storm.zookeeper.port 选项.

  1. storm.local.dir: Nimbus 和 Supervisor 守护进程需要配置一个本地目录来存储少量状态信息(例如jars包,配置文件等等). 您应该在每个机器上创建该目录,给予适当的权限,然后使用此配置填写目录位置. 例如:
storm.local.dir: "/mnt/storm"

如果您在windows下运行Strom,应该如下: yaml storm.local.dir: "C:\\storm-local" 如果您使用相对路径,那么路径是相对于(STORM_HOME). 您也可以使用默认值 $STORM_HOME/storm-local

  1. nimbus.seeds: worker节点需要知道哪些机器是主机的候选者,以便下载 topology jar和confs(nimbus.host 在1.0之后已经废弃,这里实现了HA). 例如:
nimbus.seeds: ["111.222.333.44"]

鼓励您填写机器的FQDN (Fully Qualified Domain Name,全域名)列表. 如果要设置Nimbus HA,则必须解决运行nimbus的所有机器的FQDN.当您只想设置“伪分布式”集群时您可能希望将其保留为默认值,仍然鼓励您填写FQDN.

  1. supervisor.slots.ports: 对于每个worker节点,您可以使用此配置设置在该计算机上运行的worker数量. 每个worker使用单个端口接收消息,并且此设置定义哪些端口打开以供使用. 如果您在此定义五个端口,那么Storm将分配最多五个worker在本机上运行. 如果您定义了三个端口,Storm将只能运行三个worker. 默认情况下,此设置被配置为在端口6700,6701,6702和6703上运行4个worker:
supervisor.slots.ports:
    - 6700
    - 6701
    - 6702
    - 6703 

以上是一些配置的介绍,下面我们的具体配置如下:

 storm.zookeeper.servers:
    - "hadoop000"
 storm.local.dir: "/home/hadoop/app/storm-2.1.0/data"
 nimbus.seeds: ["hadoop000"]
 storm.zookeeper.port: 2181
 supervisor.slots.ports:
    - 6700
    - 6701
    - 6702
    - 6703
 ui.port: 8082

注意:配置storm.zookeeper.servers前面有空格,等等细节很严格。另外ui服务进程的端口8080由于很容易和其他服务冲突,这里改成了8082

启动Storm进程

  • 主节点启动nimbus服务

    启动Nimbus。Run the command bin/storm nimbus under supervision on the master machine.

    storm nimbus &
  • 主节点启动UI服务:

    Run the Storm UI (a site you can access from the browser that gives diagnostics on the cluster and topologies) by running the command “bin/storm ui” under supervision. The UI can be accessed by navigating your web browser to http://{ui host}:8080. by default

    但是由于之前在配置文件中已经将UI端口修改成8082(避免和tomcat的8080冲突), 因此现在的访问方式是http://{ui host}:8082

    bin/storm ui &
  • 在从节点(工作节点)上启动supervisor服务(只不过目前是配置的伪分布式,所有服务都是在一个节点上,没有备用节点):

    storm supervisor &
  • 用jps判断是否启动成功(如果失败,则检查日志)

    img

  • 避免打印信息或者报错信息出现在屏幕上

    为了避免打印信息或者报错信息出现的屏幕上,也可以这样

    bin/storm ui  >/dev/null 2>&1 &  

    其中,2>&1 是将标准出错重定向到标准输出,但是最好的方式是采用nohup方式运行。

    nohup命令:如果你正在运行一个进程,而且你觉得在退出帐户时该进程还不会结束,那么可以使用nohup命令。该命令可以在你退出帐户/关闭终端之后继续运行相应的进程。nohup就是不挂起的意思( no hang up), 像这样:

    nohup ./bin/storm nimbus  > /dev/null  2>&1 & 
    nohup ./bin/storm supervisor > /dev/null 2>&1 & 
    nohup ./bin/storm ui > /dev/null  2>&1 &

查看UI界面

访问http://hadoop000:8082/ (端口是在配置文件中配置过的)

img

img

img

Numbus的配置超级繁杂, 有14页之多

查看log

log日志在storm包的logs文件夹中

如果想要在UI界面点击主机名或者端口号查看日志,则对应的机器需要开启logviewer服务

storm logviewer &

配置集群

假设当前机器hadoop001作为主节点, 如果要增加从节点机器hadoop002和hadoop003`,只需要修改如下:

storm.zookeeper.servers:
     - "hadoop001"
     - "hadoop002"
     - "hadoop003"
 storm.local.dir: "/home/hadoop/app/storm-2.1.0/data"
 nimbus.seeds: ["hadoop001"]
 storm.zookeeper.port: 2181
 supervisor.slots.ports:
   - 6700
   - 6701
   - 6702
   - 6703
 ui.port: 8082  

给bin目录的文件添加可执行权限:

bin>   chmod u+x *

开启nimbus和ui后台进程,把整个安装文件夹使用scp命令复制到其他两台机器(需要提前配置好通信机器之间的ssh免密访问),清空logsdata文件夹里面的内容,给bin目录的文件添加可执行权限,开启supervisor进程,这样就准备好集群环境了。

打开UI界面查看 可以看到Nimbus是hadoop001, 而supervisor是hadoop002和hadoop003

Storm UI

Cluster Summary

Version Supervisors Used slots Free slots Total slots Executors Tasks
2.1.0 2 0 8 8 0 0

Nimbus Summary

Search:

Host Port Status Version Uptime
hadoop001 6627 Leader 2.1.0 9m 16s

Showing 1 to 1 of 1 entries

Owner Summary

Search:

Owner Total Topologies Total Executors Total Workers Memory Usage (MB)
No data available in table

Showing 0 to 0 of 0 entries

Topology Summary

Search:

Name Owner Status Uptime Num workers Num executors Num tasks Replication count Assigned Mem (MB) Scheduler Info Topology Version Storm Version
No data available in table

Showing 0 to 0 of 0 entries

Supervisor Summary

Search:

Host Id Uptime Slots Used slots Avail slots Used Mem (MB) Version
hadoop002 (log) d8412147-0dbb-4933-a2d3-1bb28c806dc2-192.168.186.102 10m 33s 4 0 4 0 2.1.0
hadoop003 (log) 57798559-c338-4844-9e29-eb9aa99fa2fe-192.168.186.103 1m 7s 4 0 4 0 2.1.0

Showing 1 to 2 of 2 entries

Nimbus Configuration

Show entries

Search:

Key Value
blacklist.scheduler.reporter "org.apache.storm.scheduler.blacklist.reporters.LogReporter"
blacklist.scheduler.resume.time.secs 1800
blacklist.scheduler.strategy "org.apache.storm.scheduler.blacklist.strategies.DefaultBlacklistStrategy"
blacklist.scheduler.tolerance.count 3
blacklist.scheduler.tolerance.time.secs 300
client.blobstore.class "org.apache.storm.blobstore.NimbusBlobStore"
dev.zookeeper.path "/tmp/dev-storm-zookeeper"
drpc.authorizer.acl.filename "drpc-auth-acl.yaml"
drpc.authorizer.acl.strict false
drpc.childopts "-Xmx768m"
drpc.disable.http.binding true
drpc.http.creds.plugin "org.apache.storm.security.auth.DefaultHttpCredentialsPlugin"
drpc.http.port 3774
drpc.https.keystore.password ""
drpc.https.keystore.type "JKS"
drpc.https.port -1
drpc.invocations.port 3773
drpc.invocations.threads 64
drpc.max_buffer_size 1048576
drpc.port 3772

Showing 1 to 20 of 266 entries

为nimbus配置高可用

最后, 尽管前面我们使用了三台机器搭建了集群,由于nimbus服务没有配置高可用, 实际上还是算伪分布式。为nimbus配置高可用也非常简单, 比如说让hadoop001hadoop002其中一个作为主节点,另一个作为备用主节点,则配置如下:

nimbus.seeds: ["hadoop001", "hadoop002"]

Nimbus Summary

Search:

Host Port Status Version Uptime
hadoop001 6627 Leader 2.1.0 9m 16s
hadoop002 6627 Follower 2.1.0 ...

作业

使用Centos7系统搭建Storm集群, 将搭建过程形成笔记,并附上截图。

练习

在互联网上查询Twitter是如何使用storm进行情感分析的,了解storm的重要性以及用途。

Views: 895

1- 初识Storm

什么是Storm?

Storm为分布式实时计算提供了一组通用原语,可被用于“流处理”之中,实时处理消息并更新数据库。这是管理队列及工作者集群的另一种方式。 Storm也可被用于“连续计算”(continuous computation),对数据流做连续查询,在计算时就将结果以流的形式输出给用户。它还可被用于“分布式RPC”,以并行的方式计算。

Storm可以方便地在一个计算机集群中编写与扩展复杂的实时计算,Storm用于实时处理,就好比 Hadoop 用于批处理。Storm保证每个消息都会得到处理,而且它很快——在一个小集群中,每秒可以处理数以百万计的消息。更棒的是你可以使用任意编程语言来做开发。

离线计算和流式计算

离线计算

  • 离线计算:批量获取数据、批量传输数据、周期性批量计算数据、数据展示

  • 代表技术:Sqoop批量导入数据、HDFS批量存储数据、MapReduce批量计算、Hive

流式计算

(终极目的:留住用户,提升用户体验)

  • 流式计算:数据实时产生、数据实时传输、数据实时计算、实时展示

  • 代表技术:Flume实时获取数据、Kafka实时数据存储、Storm实时数据计算、Redis实时结果缓存。

  • 一句话总结:将源源不断产生的数据实时收集并实时计算,尽可能快的得到计算结果

image-20210830005200380

Storm与Hadoop的区别

Storm Hadoop
Storm用于实时计算 Hadoop用于离线计算
Storm的数据保存在内存,源源不断 Hadoop的数据保存在文件系统中,一批一批
Storm的数据通过网络传输进来 Hadoop的数据保存在磁盘中
Storm与Hadoop的编程模型相似 Storm与Hadoop的编程模型相似

Storm的体系结构

img

http://blog.chinaunix.net/attachment/201309/13/790245_13790273614l7w.jpg

  • Nimbus:负责资源分配和任务调度。

  • Supervisor:负责接受nimbus分配的任务,启动和停止属于自己管理的worker进程。通过配置文件设置当前supervisor上启动多少个worker。

  • Worker:运行具体处理组件逻辑的进程。Worker运行的任务类型只有两种,一种是Spout任务,一种是Bolt任务。

  • Executor:Storm 0.8之后,Executor为Worker进程中的具体的物理线程,同一个Spout/Bolt的Task可能会共享一个物理线程,一个Executor中只能运行隶属于同一个Spout/Bolt的Task。

  • Task:worker中每一个spout/bolt的线程称为一个task. 在storm0.8之后,task不再与物理线程对应,不同spout/bolt的task可能会共享一个物理线程,该线程称为executor。

storm 简介及单机版安装指南

Storm的运行机制

img

  • 整个处理流程的组织协调不用用户去关心,用户只需要去定义每一个步骤中的具体业务处理逻辑

  • 具体执行任务的角色是Worker,Worker执行任务时具体的行为则有我们定义的业务逻辑决定

img

参考

官方网站: http://storm.apache.org/
官方文档: http://storm.apache.org/releases/2.1.0/index.html

Views: 274