生产实践Kafka与ELK

简介: 项目背景,第三方发送数据到kafka。我方负责消费,解码、存储、入库。开发环境:集群1.0kafka,springboot开发程序问题:1.kafka服务端与客户端版本不一致,服务端1.0,客户端0.8,导致程序消费到一部分重启后,为消费的数据丢失直接别为已消费,导致剩余数据未走流程。

项目背景,第三方发送数据到kafka。我方负责消费,解码、存储、入库。

开发环境:集群1.0kafka,springboot开发程序

问题:1.kafka服务端与客户端版本不一致,服务端1.0,客户端0.8,导致程序消费到一部分重启后,为消费的数据丢失直接别为已消费,导致剩余数据未走流程。

2. 数据丢失节点盲目,无法找到,需要大量测试。

3. 线程数太多,分配资源浪费,dfs单节点瓶颈。


解决问题:

1. 服务端与客户端版本不一致,0.8版本中auto.offset.reset=smallest/largest/anything else  与1.0版本auto.offset.reset=earliest/latest/none

参数不一致,导致重启程序后数据丢失,调整版本一致后,解决此问题。

2. 使用springboot 继承metrics做统计

pom中添加

<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-web</artifactId>
		</dependency>
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-actuator</artifactId>
		</dependency>


代码中注入


 @Autowired
 CounterService counterService;//计数
    
 @Autowired
 GaugeService gaugeService;//计时



counterService.increment("zhibiaojishu");//计数的指标名称
gaugeService.submit("zhibiaojishi");//计时的指标名称

继承es,将指标存入到es中,

中程序中加入

@SpringBootApplication
public class KafkaApplication {

	@Bean
	@ConfigurationProperties("metrics.export")
	@ExportMetricWriter
	public MetricWriter metricWriter() {
		return new ElasticsearchMetricWriter();
	}
	public static void main(String[] args) {
		SpringApplication.run(KafkaApplication.class, args);
	}
}

具体代码可参考https://github.com/lane-cn/spring-boot-metrics-sample

安装elasticsearrch-head-1.0.0.jar      head工具 直接java -jar elasticsearrch-head-1.0.0.jar启动,配置ip和端口号

 elasticsearrch-5.6.9.tar.gz   更改集群名字和端口号  ./elasticsearrch -d 后台启动,使用内存可使用大一点

安装kibana-5.6.9.tar.gz 配置ip,启动后可收集es中的数据做统计报表,此时各项指标脚本监控形成图标已完成。


3. 线程数根据cpu物理内存做修改一般保持一致,dfs单点瓶颈问题需要搭建集群环境。


到此ELk模拟环境完成。运维人员可直接查看kibana图标监控各个数据指标。








相关文章
|
1月前
|
消息中间件 存储 监控
构建高可用性Apache Kafka集群:从理论到实践
【10月更文挑战第24天】随着大数据时代的到来,数据传输与处理的需求日益增长。Apache Kafka作为一个高性能的消息队列服务,因其出色的吞吐量、可扩展性和容错能力而受到广泛欢迎。然而,在构建大规模生产环境下的Kafka集群时,保证其高可用性是至关重要的。本文将从个人实践经验出发,详细介绍如何构建一个高可用性的Kafka集群,包括集群规划、节点配置以及故障恢复机制等方面。
78 4
|
4月前
|
分布式计算 搜索推荐 物联网
大数据及AI典型场景实践问题之通过KafKa+OTS+MaxCompute完成物联网系统技术重构如何解决
大数据及AI典型场景实践问题之通过KafKa+OTS+MaxCompute完成物联网系统技术重构如何解决
|
4月前
|
消息中间件 负载均衡 Kafka
Kafka 实现负载均衡与故障转移:深入分析 Kafka 的架构特点与实践
【8月更文挑战第24天】Apache Kafka是一款专为实时数据处理和流传输设计的高性能消息系统。其核心设计注重高吞吐量、低延迟与可扩展性,并具备出色的容错能力。Kafka采用分布式日志概念,通过数据分区及副本机制确保数据可靠性和持久性。系统包含Producer(消息生产者)、Consumer(消息消费者)和Broker(消息服务器)三大组件。Kafka利用独特的分区机制实现负载均衡,每个Topic可以被划分为多个分区,每个分区可以被复制到多个Broker上,确保数据的高可用性和可靠性。
86 2
|
4月前
|
消息中间件 存储 算法
时间轮在Kafka的实践:技术深度剖析
【8月更文挑战第13天】在分布式消息系统Kafka中,时间轮(Timing Wheel)作为一种高效的时间调度机制,被广泛应用于处理各种延时操作,如延时生产、延时拉取和延时删除等。本文将深入探讨时间轮在Kafka中的实践应用,解析其技术原理、优势及具体实现方式。
138 2
|
4月前
|
消息中间件 安全 Kafka
"深入实践Kafka多线程Consumer:案例分析、实现方式、优缺点及高效数据处理策略"
【8月更文挑战第10天】Apache Kafka是一款高性能的分布式流处理平台,以高吞吐量和可扩展性著称。为提升数据处理效率,常采用多线程消费Kafka数据。本文通过电商订单系统的案例,探讨了多线程Consumer的实现方法及其利弊,并提供示例代码。案例展示了如何通过并行处理加快订单数据的处理速度,确保数据正确性和顺序性的同时最大化资源利用。多线程Consumer有两种主要模式:每线程一个实例和单实例多worker线程。前者简单易行但资源消耗较大;后者虽能解耦消息获取与处理,却增加了系统复杂度。通过合理设计,多线程Consumer能够有效支持高并发数据处理需求。
188 4
|
5月前
|
消息中间件 分布式计算 NoSQL
EMR-Kafka Connect:高效数据迁移的革新实践与应用探索
Kafka Connect是Kafka官方提供的一个可扩展的数据传输框架,它允许用户以声明式的方式在Kafka与其他数据源之间进行数据迁移,无需编写复杂的数据传输代码。
|
5月前
|
消息中间件 Java Kafka
kafka Linux环境搭建安装及命令创建队列生产消费消息
kafka Linux环境搭建安装及命令创建队列生产消费消息
114 4
|
4月前
|
消息中间件 Kafka
一文吃透企业级elk技术栈:4. kafka 集群部署
一文吃透企业级elk技术栈:4. kafka 集群部署
|
5月前
|
消息中间件 存储 缓存
面试题Kafka问题之Kafka的生产消费基本流程如何解决
面试题Kafka问题之Kafka的生产消费基本流程如何解决
47 1
|
5月前
|
消息中间件 存储 Kafka
微服务分布问题之Kafka分区的副本和分布如何解决
微服务分布问题之Kafka分区的副本和分布如何解决