kafka 主动消费_SpringBoot2 整合Kafka组件,应用案例和流程详解
本文源码:Git || Gitee
一、搭建Kafka环境
1、下载解压
-- 下载wget http://mirror.bit.edu.cn/apache/kafka/2.2.0/kafka_2.11-2.2.0.tgz-- 解压tar -zxvf kafka_2.11-2.2.0.tgz-- 重命名mv kafka_2.11-2.2.0 kafka2.11
2、启动Kafka服务
kafka依赖ZooKeeper服务,需要本地安装并启动ZooKeeper。文章末尾有安装流程。
-- 执行位置-- /usr/local/mysoft/kafka2.11bin/kafka-server-start.sh config/server.properties
3、查看服务
ps -aux |grep kafka
4、开放地址端口
-- 基础路径-- /usr/local/mysoft/kafka2.11/configvim server.properties-- 添加下面注释advertised.listeners=PLAINTEXT://192.168.72.130:9092
二、Kafka基础概念
1、基础描述
Kafka是由Apache开源,具有分布式、分区的、多副本的、多订阅者,基于Zookeeper协调的分布式处理平台,由Scala和Java语言编写。通常用来搜集用户在应用服务中产生的动作日志数据,并高速的处理。日志类的数据需要高吞吐量的性能要求,对于像Hadoop一样的日志数据和离线分析系统,但又要求实时处理的限制,这是一个可行的解决方案。Kafka的目的是通过Hadoop的并行加载机制来统一线上和离线的消息处理,也是为了通过集群来提供实时的消息。
2、功能特点
(1)、通过磁盘数据结构提供消息的持久化,消息存储也能够保持长时间稳定性;
(2)、高吞吐量,即使是非常普通的硬件Kafka也可以支持每秒超高的并发量;
(3)、支持通过Kafka服务器和消费机集群来分区消息;
(4)、支持Hadoop并行数据加载;
(5)、API包封装的非常好,简单易用,上手快 ;
(6)、分布式消息队列。Kafka对消息保存时根据Topic进行归类,发送消息者称为Producer,消息接受者称为Consumer;
3、消息功能
- 点对点模式
点对点模型通常是一个基于拉取或者轮询的消息传递模型,消费者主动拉取数据,消息收到后从队列移除消息,这种模型不是将消息推送到客户端,而是从队列中请求消息。特点是发送到队列的消息被一个且只有一个消费者接收处理,即使有多个消费者监听队列也是如此。
- 发布订阅模式
发布订阅模型则是一个基于推送的消息传送模型,消息产生后,推送给所有订阅者。发布订阅模型可以有多种不同的订阅者,临时订阅者只在主动监听主题时才接收消息,而持久订阅者则监听主题的所有消息,即使当前订阅者不可用,处于离线状态。
4、消息队列作用
- 程序解耦,生产者和消费者独立,各自异步执行;
- 消息数据进行持久化存储,直到被全部消费,规避了数据丢失风险;
- 流量削峰,使用消息队列承接访问压力,尽量避免程序雪崩 ;
- 降低进程间的耦合度,系统部分组件崩溃时,不会影响到整个系统;
- 保证消息顺序执行,解决特定场景业务需求 ;
5、专业术语简介
- Broker
一台kafka服务器就是一个broker。一个集群由多个broker组成。一个broker可以容纳多个topic。
- Producer
消息生产者,就是向kafka broker发消息的客户端。
- Consumer
消息消费者,向kafka broker取消息的客户端。
- Topic
每条发布到Kafka集群的消息都有一个类别,这个类别被称为Topic,可以理解为一个队列。
- Consumer Group
每个Consumer属于一个特定的Consumer Group,可为每个Consumer指定group name,若不指定group name则属于默认的分组。
- Partition
一个庞大大的topic可以分布到多个broker上,一个topic可以分为多个partition,每个partition是一个有序的队列。partition中的每条消息都会被分配一个有序的id。kafka只保证按一个partition中的顺序将消息发给consumer,不保证一个topic的整体的顺序。Partition是物理上的概念,方便在集群中扩展,提高并发。
三、整合SpringBoot2框架
1、案例结构
- 消息生产者 : kafka-producer-server
- 消息消费方 : kafka-consumer-server
2、基础依赖
org.springframework.boot spring-boot-starter-weborg.springframework.kafka spring-kafka 2.2.4.RELEASE
3、生产者配置
spring: kafka: bootstrap-servers: 127.0.0.1:9092
4、消息生成
@RestControllerpublic class ProducerWeb { @Resource private KafkaTemplate kafkaTemplate; @RequestMapping("/send") public String sendMsg () { MsgLog msgLog = new MsgLog(1,"消息生成", 1,"消息日志",new Date()) ; String msg = JSON.toJSONString(msgLog) ; // 这里Topic如果不存在,会自动创建 kafkaTemplate.send("cicada-topic", msg); return msg ; }}
5、消费者配置
spring: kafka: bootstrap-servers: 127.0.0.1:9092 consumer: group-id: test-consumer-group
6、消息消费
@Componentpublic class ConsumerMsg { private static Logger LOGGER = LoggerFactory.getLogger(ConsumerMsg.class); @KafkaListener(topics = "cicada-topic") public void listenMsg (ConsumerRecord,String> record) { String value = record.value(); LOGGER.info("ConsumerMsg====>>"+value); }}
四、消息流程分析
1、生产者分析
- 写入方式
生产者基于推push推模式将消息发布到broker,每条消息都被追加到分区patition中,属于磁盘顺序写,效率比随机写内存要高,保障kafka高吞吐量。
- 分区概念
消息发送时都被发送到一个topic,而topic是由Partition Logs(分区日志)组成,其组织结构如下图所示:
每个Partition中的消息都是有序的,生产的消息被不断追加到Partitionlog上,其中的每一个消息都被赋予了一个唯一的offset值。每个Partition可以通过调整以适配它所在的机器,而一个topic又可以有多个Partition组成,因此整个集群就可以适应任意大小的数据。分区的原则:指定patition,则直接使用;未指定patition但指定key,通过对key的value进行hash出一个patition;patition和key都未指定,使用轮询选出一个patition。
2、消费者分析
- 消费图解
消费者是以consumer group消费者组的方式工作,由一个或者多个消费者组成一个组,共同消费一个topic。每个分区在同一时间只能由group中的一个消费者读取,但是多个group可以同时消费一个partition。
- 消费方式
消费者采用pull拉模式从broker中读取数据。对于Kafka而言,pull模式更合适,它可简化broker的设计,consumer可自主控制消费消息的速率,同时consumer可以自己控制消费方式——即可批量消费也可逐条消费,同时还能选择不同的提交方式从而实现不同的数据传输场景。
推荐阅读:
kafka 主动消费_SpringBoot2 整合Kafka组件,应用案例和流程详解相关推荐
- SpringBoot2 整合Kafka组件,应用案例和流程详解
本文源码:GitHub·点这里 || GitEE·点这里 一.搭建Kafka环境 1.下载解压 -- 下载 wget http://mirror.bit.edu.cn/apache/kafka/2.2 ...
- php事件和行为,Yii框架组件和事件行为管理详解
Yii框架组件和事件行为管理详解 来源:中文源码网 浏览: 次 日期:2018年9月2日 [下载文档: Yii框架组件和事件行为管理详解.txt ] (友情提示:右键点上行txt文档名- ...
- vue 子级拿值_vue 父组件通过$refs获取子组件的值和方法详解
前言 在vue项目中组件之间的通讯是很常见的问题,同时也是很重要的问题,我们大致可以将其分为三种情况: 父传子:在父组件中绑定值,在子组件中用props接收 子传父:在父组件中监听一个事件,在子组件中 ...
- mvc jquery ajax分页实例,jQuery Ajax自定义分页组件(jquery.loehpagerv1.0)实例详解,mvcpagerajax分页...
jQuery Ajax自定义分页组件(jquery.loehpagerv1.0)实例详解,mvcpagerajax分页 简单的两个步骤即可实现分页功能 //回调里面进行业务处理 function lo ...
- Spring Boot 整合 shiro 之盐值加密认证详解(六)
Spring Boot 整合 shiro 之盐值加密认证详解 概述 不加盐认证 加入密码认证核心代码 修改 CustomRealm 新增获取密文的方法 修改 doGetAuthenticationIn ...
- kafka 主动消费_Kafka消费组(consumer group)
在开始之前,我想花一点时间先来明确一些概念和术语,这会极大地方便我们下面的讨论.另外请原谅这文章有点长,毕竟要讨论的东西很多,虽然已然删除了很多太过细节的东西. 一. 误区澄清与概念明确 1 Kafk ...
- kafka 主动消费_Kafka核心API——Consumer消费者
Consumer之自动提交 在上文中介绍了Producer API的使用,现在我们已经知道如何将消息通过API发送到Kafka中了,那么现在的生产者/消费者模型就还差一位扮演消费者的角色了.因此,本文 ...
- kafka 安装使用 /springboot整合kafka /消息投递机制以及存储策略 /副本处理机制
一.背景 1.基本信息 Kafka是由Apache软件基金会开发的一个开源流处理平台,由Scala和Java编写.Kafka是一种高吞吐量的分布式发布订阅消息系统,它可以处理消费者在网站中的所有动作流 ...
- Kafka原理以及SpringBoot整合Kafka
1.Kafka原理 1. brokers有多个broker组成,broker是指Kafka服务器(192.168.223.140就是其中的一个broker),上面三台Kafka服务器组成了Kafka集 ...
最新文章
- [转]使用 .NET Framework 2.0 在您的应用程序中支持证书
- 机器人铁锈斑斑好吗_工业机器人系统运维员是什么职业?这个职业好吗?去哪里学?...
- Unity3D研究院之获取摄像机的视口区域
- 安卓逆向_10 --- Log 日志的插入和分析、toast方法、栈跟踪
- 学学Nhibernate
- C# WPF ASP.net 上传多文件和数据
- EXP-00091 Exporting questionable statistics
- 计算机二级python看什么书好_2021年全国计算机二级Python备考指导
- linux中通过关键字查询日志,查询结果突显
- 从实战经验来看 究竟如何才能做出赚钱的量化投资策略?
- 文本处理命令 cat more less cut wc sort uniq
- 正则表达式转NFA,DFA,最小化DFA
- mysql练习-数据查询之嵌套查询
- 牛客网-华为机试练习题 16 购物单
- python读取图像灰度值_python 读取灰度图像Python字符串处理
- 考研复试(控制工程专硕)及大学本科(物联网工程)知识点回顾(五)——其他重点内容
- 如何办理《网络出版服务许可证》
- Ahmedabad公司的LiFi技术为Nadabet提供互联网接入能力
- elasticsearch(ES)的安装部署及其插件安装
- 即时通讯(IM)开源项目OpenIM重构版本发布- v2.0.0