SSM(十六) 曲线救国-Kafka消费异常

最近线上遇到一个问题:在消费kafka消息的时候如果长时间(大概半天到一天的时间)队列里没有消息就可能再也消费不了。针对这个问题我们反复调试多次。线下模拟,调整代码,但貌似还是没有找到原因。但是只要重启消费进程就又可以继续消费。

解决方案

由于线上业务非常依赖kafka的消费,但一时半会也没有找到原因,所以最后只能想一个临时的替换方案:

基于重启就可以消费这个特点,我们在每次消费的时候都记下当前的时间点,当这个时间点在十分钟之内都没有更新我们就认为当前队列中没有消息了,就需要重启下消费进程。

既然是需要重启,由于目前还没有上分布式调度中心所以需要crontab来配合调度:每隔一分钟会调用一个shell脚本,该脚本会判断当前进程是否存在,如果存在则什么都不作,不存在则启动消费进程。

具体实现

消费程序:

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

62

63

64

65

66

67

68

69

70

71

72

73

74

75

76

77

78

79

80

81

82

83

84

85

86

87

88

89

90

91

92

93

94

95

96

97

98

99

100

101

102

103

104

105

106

* kafka消费

*

* @author crossoverJie

* @date 2017年6月19日 下午3:15:16

*/

public class  {

private static final Logger LOGGER = LoggerFactory.getLogger(KafkaMsgConsumer.class);

private static final int CORE_POOL_SIZE = 4;

private static final int MAXIMUM_POOL_SIZE = 4;

private static final int BLOCKING_QUEUE_CAPACITY = 4000;

private static final String KAFKA_CONFIG = "kafkaConfig";

private static final ExecutorService fixedThreadPool = new ThreadPoolExecutor(CORE_POOL_SIZE, MAXIMUM_POOL_SIZE, 0L, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<Runnable>(BLOCKING_QUEUE_CAPACITY));

//最后更新时间

private static AtomicLong LAST_MESSAGE_TIME = new AtomicLong(DateUtil.getLongTime());

private static MsgIterator iter = null;

private static String topic;//主题名称

static {

Properties properties = new Properties();

String path = System.getProperty(KAFKA_CONFIG);

checkArguments(!StringUtils.isBlank(path), "启动参数中没有配置kafka_easyframe_msg参数来指定kafka启动参数,请使用-DkafkaConfig=/path/fileName/easyframe-msg.properties");

try {

properties.load(new FileInputStream(new File(path)));

} catch (IOException e) {

LOGGER.error("IOException" ,e);

}

EasyMsgConfig.setProperties(properties);

}

private static void iteratorTopic() {

if (iter == null) {

iter = MsgUtil.consume(topic);

}

long i = 0L;

while (iter.hasNext()) {

i++;

if (i % 10000 == 0) {

LOGGER.info("consume i:" + i);

}

try {

String message = iter.next();

if (StringUtils.isEmpty(message)) {

continue;

}

LAST_MESSAGE_TIME = new AtomicLong(DateUtil.getLongTime());

//处理消息

LOGGER.debug("msg = " + JSON.toJSONString(message));

} catch (Exception e) {

LOGGER.error("KafkaMsgConsumer err:", e);

try {

Thread.sleep(1000);

} catch (InterruptedException e1) {

LOGGER.error("Thread InterruptedException", e1);

}

break;

}

}

}

public static void main(String[] args) {

topic = System.getProperty("topic");

checkArguments(!StringUtils.isBlank(topic), "system property topic or log_path is must!");

while (true) {

try {

iteratorTopic();

} catch (Exception e) {

MsgUtil.shutdownConsummer();

iter = null;

LOGGER.error("KafkaMsgConsumer err:", e);

try {

Thread.sleep(1000);

} catch (InterruptedException e1) {

LOGGER.error("Thread InterruptedException", e1);

}

} finally {

//此处关闭之后,由crontab每分钟检查一次,挂掉的话会重新拉起来

if (DateUtil.getLongTime() - LAST_MESSAGE_TIME.get() > 10 * 60) { //10分钟

fixedThreadPool.shutdown();

LOGGER.info("线程池是否关闭:" + fixedThreadPool.isShutdown());

try {

//当前线程阻塞10ms后,去检测线程池是否终止,终止则返回true

while (!fixedThreadPool.awaitTermination(10, TimeUnit.MILLISECONDS)) {

LOGGER.info("检测线程池是否终止:" + fixedThreadPool.isTerminated());

}

} catch (InterruptedException e) {

LOGGER.error("等待线程池关闭错误", e);

}

LOGGER.info("线程池是否终止:" + fixedThreadPool.isTerminated());

LOGGER.info("in 10 min dont have data break");

break;

}

}

}

LOGGER.info("app shutdown");

System.exit(0);

}

}

在线代码

需要配合以下这个shell脚本运行:

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20
#!/bin/sh

#crontab

# * * * * * sh /data/schedule/kafka/run-kafka-consumer.sh >>/data/schedule/kafka/run-sms-log.log

# 如果进程存在就不启动

a1=`ps -ef|grep 'KafkaMsgConsumer'|grep -v grep|wc -l`

if [ $a1 -gt 0  ];then

echo "=======     `date +'%Y-%m-%d %H:%M:%S'` KafkaMsgConsumer  is EXIT...=======     "

exit

fi

LANG="zh_CN.UTF-8"

nohup /opt/java/jdk1.7.0_80/bin/java -d64 -Djava.security.egd=file:/dev/./urandom

-Djava.ext.dirs=/opt/tomcat/webapps/ROOT/WEB-INF/lib

-Dtopic=TOPIC_A

-Dlogback.configurationFile=/data/schedule/kafka/logback.xml

-DkafkaConfig=/opt/tomcat/iopconf/easyframe-msg.properties

-classpath /opt/tomcat/webapps/ROOT/WEB-INF/classes com.crossoverJie.kafka.SMSMsgConsumer >> /data/schedule/kafka/smslog/kafka.log 2>&1 &

echo "`date +'%Y-%m-%d %H:%M:%S'`  KafkaMsgConsumer running...."

在线代码

再配合crontab的调度:

1
* * * * * sh /data/schedule/kafka/run-kafka-consumer.sh >>/data/schedule/kafka/run-sms-log.log

即可。

总结

虽说处理起来很简单,但依然是治标不治本,依赖的东西比较多(shell脚本,调度)。
所以也问问各位有没有什么思路:

生产配置:

  • 三台kafka、ZK组成的集群。

其中也有其他团队的消费程序在正常运行,应该和kafka的配置没有关系。

项目地址:https://github.com/crossoverJie/SSM.git

个人博客:http://crossoverjie.top

原文:大专栏  SSM(十六) 曲线救国-Kafka消费异常

原文地址:https://www.cnblogs.com/petewell/p/11612093.html

时间: 2024-11-05 22:36:08

SSM(十六) 曲线救国-Kafka消费异常的相关文章

【读书笔记】C#高级编程 第十六章 错误和异常

(一)简介 错误的出现并不总是编写应用程序的人的原因,有时应用程序会因为应用程序的最终用户引发或运行代码的环境而发生错误.C#提供了异常处理机制来处理错误. (二)异常类 在C#中,但刚出现某个特殊的异常错误条件时,就会创建(或抛出)一个异常对象.一般情况下异常没有特定的名称空间,异常类应放在生成异常的类所在的名称空间. (三)捕获异常 .NET Framework提供了大量的预定义基类异常对象.为了在C#代码中处理可能的错误情况,一般要把程序的相关部分分成3种不同类型的代码块. try块包含的

企业搜索引擎开发之连接器connector(二十六)

连接器通过监视器对象DocumentSnapshotRepositoryMonitor从上文提到的仓库对象SnapshotRepository(数据库仓库为DBSnapshotRepository)中迭代获取数据 监视器类DocumentSnapshotRepositoryMonitor在其构造方法初始化相关成员变量,这些成员属性都是与数据获取及数据处理逻辑相关的对象 /** This connector instance's current traversal schedule. */ pri

Java Web总结十六之一自定义标签

一.自定义标签简介 1.为什么要使用自定义标签? 自定义标签主要用于移除Jsp页面中的<%java%>代码. 2.开发自定义标签的步骤: 1)开发自定义标签处理类,该类需要实现SimpleTag接口/SimpleTagSupport类,重写doTag()方法. 2)编写标签库描述符(tld)文件,在tld文件中对自定义标签进行描述,并放置在WEB-INF/目录下. 3)完成以上操作,即可在JSP页面中导入和使用自定义标签. 二.自定义标签描述 1.实现SimpleTag接口的标签通常称为简单标

Spring+SpringMVC+MyBatis深入学习及搭建(十六)——SpringMVC注解开发(高级篇)

转载请注明出处:http://www.cnblogs.com/Joanna-Yan/p/7085268.html 前面讲到:Spring+SpringMVC+MyBatis深入学习及搭建(十五)——SpringMVC注解开发(基础篇) 本文主要内容: (1)SpringMVC校验 (2)数据回显 (3)异常处理器 (4)图片上传 (5)Json数据交互 (6)支持RESTful 1.SpringMVC校验 1.1校验理解 项目中,通常使用较多的是前端的校验,比如页面中js校验.对于安全要求较高的

十六、变更管理; 十七、信息系统安全管理; 十八、项目风险管理

十六.变更管理 1.  变更的工作程序 (1).提出与接受变更申请 (2).对变更的初审 (3).变更方案论证 (4).项目变更控制委员会审查 (5).发出变更通知并开始实施 (6).变更实施的监控 (7).变更效果的评估 (8).判断发生变更后的项目是否已纳入正常轨道 2.  变更初审的4条内容 (1).对变更提出方施加影响,确认变更的必要性,确保变更是有价值的 (2).格式校验,完整性较验,确保评估所需信息准备充分 (3).在干系人间就提出供评估的变更信息达成共识 (4).变更初审的常见方式

SSM(十四) 基于annotation的http防重插件

SSM(十四) 基于annotation的http防重插件 前言 针对于我们现在常用的RESTful API通常我们需要对请求进行唯一标识,也就是每次都要带上一个请求号,如reqNO. 对于入库这种操作数据库的请求我们一般要保证他的唯一性,一个请求号通常只能用一次,所以需要我们对这种请求加上校验机制. 该需求的实现思路是通过自定义annotation,只给需要进行校验的接口加上注解.然后通过切面使用了注解的接口将每次请求号存进Redis,每次都进行判断是否存在这个请求号即可. 来看下加上本次插件

SSM(十五) 乐观锁与悲观锁的实际应用

SSM(十五) 乐观锁与悲观锁的实际应用 前言 随着互联网的兴起,现在三高(高可用.高性能.高并发)项目是越来越流行. 本次来谈谈高并发.首先假设一个业务场景:数据库中有一条数据,需要获取到当前的值,在当前值的基础上+10,然后再更新回去.如果此时有两个线程同时并发处理,第一个线程拿到数据是10,+10=20更新回去.第二个线程原本是要在第一个线程的基础上再+20=40,结果由于并发访问取到更新前的数据为10,+20=30. 这就是典型的存在中间状态,导致数据不正确.来看以下的例子: 并发所带来

面向对象程序设计-C++ Finial exam review NOTES【第十六次上课笔记】

写在前面: 我记得也不全,如果有记录的更全的同学可以留言,我会添加哒 :) 常量 内敛函数 为什么需要内敛函数 内敛函数适用于什么场合 内敛函数本身,最大优点是,避免了真正函数调用的开销 因为普通函数调用会有开销,比如开辟一个栈,结束了还要释放局部变量 如果函数体只有寥寥几行,是不值得使用函数 在函数体代码比较短小的时候,使用频繁的,应该使用内敛函数 最大优点:没有函数调用开销,又解决了带有参数宏的简单替换,它有类型检查 引用 什么是引用:给这块区域的数据再增加一个名称(本质含义) 表面上看,相

kafka 消费?

前置资料  kafka kafka消费中的问题及解决方法: 情况1: 问题:脚本读取kafka 数据,写入到数据库,有时候出现MySQL server has gone away,导致脚本死掉.再次启动,这过程中的kafka数据丢失. 原因:MySQL server has gone away 出现可能是连接超时,可能超过每秒请求上限-这些异常是小概率事件,难以避免.git kafka 的demo脚本是实时监听的脚本, 简单明了,没有再去针对kafka偏移量研究:但是一旦断掉, 过程中的kafk