如何保证kafka 的消息机制 ack-fail 源码跟踪

如何保证kafka 的消息机制 ack-fail 源码跟踪,第1张

Kafka is a distributed, partitioned, replicated commit log service It provides the functionality of a messaging system, but with a unique design(Kafka布式、区(partitioned)、基于备份(replicated)commit-log存储服务提供类似于messaging system特性,设计实现完全同)kafka种高吞吐量布式发布订阅消息系统特性:
(1)、通O(1)磁盘数据结构提供消息持久化种结构于即使数TB消息存储能够保持间稳定性能
(2)、高吞吐量:即使非普通硬件kafka支持每秒数十万消息
(3)、支持通kafka服务器消费机集群区消息
(4)、支持Hadoop并行数据加载
、用Kafka面自带脚本进行编译
载Kafka源码面自带gradlew脚本我利用编译Kafka源码:
1 # wget
2 # tar -zxf kafka-0811-srctgz
3 # cd kafka-0811-src
4 # /gradlew releaseTarGz
运行面命令进行编译现异信息:
01 :core:signArchives FAILED
02
03 FAILURE: Build failed with an exception
04
05 What went wrong:
06 Execution failed for task ':core:signArchives'
07 > Cannot perform signing task ':core:signArchives' because it
08 has no configured signatory
09
10 Try:
11 Run with --stacktrace option to get the stack trace Run with
12 --info or --debug option to get more log output
13
14 BUILD FAILED
bug()用面命令进行编译
1 /gradlew releaseTarGzAll -x signArchives
候编译功(编译程现)编译程我指定应Scala版本进行编译:
1 /gradlew -PscalaVersion=2103 releaseTarGz -x signArchives
编译完core/build/distributions/面kafka_210-0811tgz文件网载直接用
二、利用sbt进行编译
我同用sbt编译Kafka步骤:
01 # git clone
02 # cd kafka
03 # git checkout -b 08 remotes/origin/08
04 # /sbt update
05 [info] [SUCCESSFUL ] orgeclipsejdt#core;311!corejar (2243ms)
06 [info] downloading
07 [info] [SUCCESSFUL ] ant#ant;165!antjar (1150ms)
08 [info] Done updating
09 [info] Resolving orgapachehadoop#hadoop-core;0202
10 [info] Done updating
11 [info] Resolving comyammermetrics#metrics-annotation;220
12 [info] Done updating
13 [info] Resolving comyammermetrics#metrics-annotation;220
14 [info] Done updating
15 [success] Total time: 168 s, completed Jun 18, 2014 6:51:38 PM
16
17 # /sbt package
18 [info] Set current project to Kafka (in build file:/export1/spark/kafka/)
19 Getting Scala 280
20 :: retrieving :: orgscala-sbt#boot-scala
21 confs: [default]
22 3 artifacts copied, 0 already retrieved (14544kB/27ms)
23 [success] Total time: 1 s, completed Jun 18, 2014 6:52:37 PM
于Kafka 08及版本需要运行命令:
01 # /sbt assembly-package-dependency
02 [info] Loading project definition from /export1/spark/kafka/project
03 [warn] Multiple resolvers having different access mechanism configured with
04 same name 'sbt-plugin-releases' To avoid conflict, Remove duplicate project
05 resolvers (`resolvers`) or rename publishing resolver (`publishTo`)
06 [info] Set current project to Kafka (in build file:/export1/spark/kafka/)
07 [warn] Credentials file /home/wyp/m2/credentials does not exist
08 [info] Including slf4j-api-172jar
09 [info] Including metrics-annotation-220jar
10 [info] Including scala-compilerjar
11 [info] Including scala-libraryjar
12 [info] Including slf4j-simple-164jar
13 [info] Including metrics-core-220jar
14 [info] Including snappy-java-1041jar
15 [info] Including zookeeper-334jar
16 [info] Including log4j-1215jar
17 [info] Including zkclient-03jar
18 [info] Including jopt-simple-32jar
19 [warn] Merging 'META-INF/NOTICE' with strategy 'rename'
20 [warn] Merging 'org/xerial/snappy/native/README' with strategy 'rename'
21 [warn] Merging 'META-INF/maven/orgxerialsnappy/snappy-java/LICENSE'
22 with strategy 'rename'
23 [warn] Merging 'LICENSEtxt' with strategy 'rename'
24 [warn] Merging 'META-INF/LICENSE' with strategy 'rename'
25 [warn] Merging 'META-INF/MANIFESTMF' with strategy 'discard'
26 [warn] Strategy 'discard' was applied to a file
27 [warn] Strategy 'rename' was applied to 5 files
28 [success] Total time: 3 s, completed Jun 18, 2014 6:53:41 PM
我sbt面指定scala版本:
01 <!--
02 User: 往记忆
03 Date: 14-6-18
04 Time: 20:20
05 bolg:
06 本文址:/archives/1044
07 往记忆博客专注于hadoop、hive、spark、shark、flume技术博客量干货
08 往记忆博客微信公共帐号:iteblog_hadoop
09 -->
10 sbt "++2103 update"
11 sbt "++2103 package"
12 sbt "++2103 assembly-package-dependency"

一、消息中间件相关知识

1、概述

消息队列已经逐渐成为企业IT系统内部通信的核心手段。它具有低耦合、可靠投递、广播、流量控制、最终一致性等一系列功能,成为异步RPC的主要手段之一。当今市面上有很多主流的消息中间件,如老牌的ActiveMQ、RabbitMQ,炙手可热的Kafka,阿里巴巴自主开发RocketMQ等。

2、消息中间件的组成

21 Broker

消息服务器,作为server提供消息核心服务

22 Producer

消息生产者,业务的发起方,负责生产消息传输给broker,

23 Consumer

消息消费者,业务的处理方,负责从broker获取消息并进行业务逻辑处理

24 Topic

25 Queue

26 Message

消息体,根据不同通信协议定义的固定格式进行编码的数据包,来封装业务数据,实现消息的传输

3 消息中间件模式分类

31 点对点

PTP点对点:使用queue作为通信载体

说明:

消息生产者生产消息发送到queue中,然后消息消费者从queue中取出并且消费消息。

消息被消费以后,queue中不再存储,所以消息消费者不可能消费到已经被消费的消息。 Queue支持存在多个消费者,但是对一个消息而言,只会有一个消费者可以消费。

说明:

queue实现了负载均衡,将producer生产的消息发送到消息队列中,由多个消费者消费。但一个消息只能被一个消费者接受,当没有消费者可用时,这个消息会被保存直到有一个可用的消费者。

4 消息中间件的优势

41 系统解耦

交互系统之间没有直接的调用关系,只是通过消息传输,故系统侵入性不强,耦合度低。

42 提高系统响应时间

例如原来的一套逻辑,完成支付可能涉及先修改订单状态、计算会员积分、通知物流配送几个逻辑才能完成;通过MQ架构设计,就可将紧急重要(需要立刻响应)的业务放到该调用方法中,响应要求不高的使用消息队列,放到MQ队列中,供消费者处理。

43 为大数据处理架构提供服务

通过消息作为整合,大数据的背景下,消息队列还与实时处理架构整合,为数据处理提供性能支持。

44 Java消息服务——JMS

Java消息服务(Java Message Service,JMS)应用程序接口是一个Java平台中关于面向消息中间件(MOM)的API,用于在两个应用程序之间,或分布式系统中发送消息,进行异步通信。

5 消息中间件应用场景

51 异步通信

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

52 解耦

降低工程间的强依赖程度,针对异构系统进行适配。在项目启动之初来预测将来项目会碰到什么需求,是极其困难的。通过消息系统在处理过程中间插入了一个隐含的、基于数据的接口层,两边的处理过程都要实现这一接口,当应用发生变化时,可以独立的扩展或修改两边的处理过程,只要确保它们遵守同样的接口约束。

53 冗余

有些情况下,处理数据的过程会失败。除非数据被持久化,否则将造成丢失。消息队列把数据进行持久化直到它们已经被完全处理,通过这一方式规避了数据丢失风险。许多消息队列所采用的”插入-获取-删除”范式中,在把一个消息从队列中删除之前,需要你的处理系统明确的指出该消息已经被处理完毕,从而确保你的数据被安全的保存直到你使用完毕。

54 扩展性

因为消息队列解耦了你的处理过程,所以增大消息入队和处理的频率是很容易的,只要另外增加处理过程即可。不需要改变代码、不需要调节参数。便于分布式扩容。

55 过载保护

在访问量剧增的情况下,应用仍然需要继续发挥作用,但是这样的突发流量无法提取预知;如果以为了能处理这类瞬间峰值访问为标准来投入资源随时待命无疑是巨大的浪费。使用消息队列能够使关键组件顶住突发的访问压力,而不会因为突发的超负荷的请求而完全崩溃。

56 可恢复性

系统的一部分组件失效时,不会影响到整个系统。消息队列降低了进程间的耦合度,所以即使一个处理消息的进程挂掉,加入队列中的消息仍然可以在系统恢复后被处理。

57 顺序保证

在大多使用场景下,数据处理的顺序都很重要。大部分消息队列本来就是排序的,并且能保证数据会按照特定的顺序来处理。

58 缓冲

在任何重要的系统中,都会有需要不同的处理时间的元素。消息队列通过一个缓冲层来帮助任务最高效率的执行,该缓冲有助于控制和优化数据流经过系统的速度。以调节系统响应时间。

59 数据流处理

分布式系统产生的海量数据流,如:业务日志、监控数据、用户行为等,针对这些数据流进行实时或批量采集汇总,然后进行大数据分析是当前互联网的必备技术,通过消息队列完成此类数据收集是最好的选择。

6 消息中间件常用协议

61 AMQP协议

AMQP即Advanced Message Queuing Protocol,一个提供统一消息服务的应用层标准高级消息队列协议,是应用层协议的一个开放标准,为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息,并不受客户端/中间件不同产品,不同开发语言等条件的限制。

优点:可靠、通用

62 MQTT协议

MQTT(Message Queuing Telemetry Transport,消息队列遥测传输)是IBM开发的一个即时通讯协议,有可能成为物联网的重要组成部分。该协议支持所有平台,几乎可以把所有联网物品和外部连接起来,被用来当做传感器和致动器(比如通过Twitter让房屋联网)的通信协议。

优点:格式简洁、占用带宽小、移动端通信、PUSH、嵌入式系统

63 STOMP协议

STOMP(Streaming Text Orientated Message Protocol)是流文本定向消息协议,是一种为MOM(Message Oriented Middleware,面向消息的中间件)设计的简单文本协议。STOMP提供一个可互 *** 作的连接格式,允许客户端与任意STOMP消息代理(Broker)进行交互。

优点:命令模式(非topic\queue模式)

64 XMPP协议

XMPP(可扩展消息处理现场协议,Extensible Messaging and Presence Protocol)是基于可扩展标记语言(XML)的协议,多用于即时消息(IM)以及在线现场探测。适用于服务器之间的准即时 *** 作。核心是基于XML流传输,这个协议可能最终允许因特网用户向因特网上的其他任何人发送即时消息,即使其 *** 作系统和浏览器不同。

优点:通用公开、兼容性强、可扩展、安全性高,但XML编码格式占用带宽大

65 其他基于TCP/IP自定义的协议

有些特殊框架(如:redis、kafka、zeroMq等)根据自身需要未严格遵循MQ规范,而是基于TCP\IP自行封装了一套协议,通过网络socket接口进行传输,实现了MQ的功能。

7 常见消息中间件MQ介绍

71 RocketMQ

阿里系下开源的一款分布式、队列模型的消息中间件,原名Metaq,30版本名称改为RocketMQ,是阿里参照kafka设计思想使用java实现的一套mq。同时将阿里系内部多款mq产品(Notify、metaq)进行整合,只维护核心功能,去除了所有其他运行时依赖,保证核心功能最简化,在此基础上配合阿里上述其他开源产品实现不同场景下mq的架构,目前主要多用于订单交易系统。

具有以下特点:

官方提供了一些不同于kafka的对比差异:

>kafka重复消费的根本原因就是“数据消费了,但是offset没更新”!而我们要探究一般什么情况下会导致offset没更新?
maxpollintervalms

两次poll *** 作允许的最大时间间隔。单位毫秒。默认值300000(5分钟)。

两次poll超过此时间间隔,Kafka服务端会进行rebalance *** 作,导致客户端连接失效,无法提交offset信息,从而引发重复消费。

拿到消息就提交offset
1、丢包问题 :消息推送服务,每天早上,手机上各终端都会给用户推送消息,这时候流量剧增,可能会出现kafka发送数据过快,导致服务器网卡爆满,或者磁盘处于繁忙状态,可能会出现丢包现象。

解决方案:首先对kafka进行限速, 其次启用重试机制,重试间隔时间设置长一些,最后Kafka设置acks=all,即需要相应的所有处于ISR的分区都确认收到该消息后,才算发送成功。 

检测方法:使用重放机制,查看问题所在。

2 重复消费最常见的原因 :re-balance问题,通常会遇到消费的数据,处理很耗时,导致超过了Kafka的session timeout时间(010x版本默认是30秒),那么就会re-balance重平衡,此时有一定几率offset没提交,会导致重平衡后重复消费。 

消息重复消费和消息丢包的解决办法

保证不丢失消息:生产者(ack=all 代表至少成功发送一次)     重试机制

消费者 (offset手动提交,业务逻辑成功处理后,提交offset) 

保证不重复消费:落表(主键或者唯一索引的方式,避免重复数据) 

业务逻辑处理(选择唯一主键存储到Redis或者mongdb中,先查询是否存在,若存在则不处理;若不存在,先插入Redis或Mongdb,再进行业务逻辑处理)


欢迎分享,转载请注明来源:内存溢出

原文地址: http://outofmemory.cn/zz/10649672.html

(0)
打赏 微信扫一扫 微信扫一扫 支付宝扫一扫 支付宝扫一扫
上一篇 2023-05-10
下一篇 2023-05-10

发表评论

登录后才能评论

评论列表(0条)

保存