消息队列之Kafka

背景:Dubbo远程调用的性能问题

Dubbo调用普遍存在于我们的微服务项目中, 这些Dubbo调用全部是同步的操作, 这里的"同步"指:消费者A调用生产者B之后,A的线程会进入阻塞状态,等待生产者B运行结束返回之后,A才能运行之后的代码, Dubbo消费者发送调用后进入阻塞状态,这个状态表示该线程仍占用内存资源,但是什么动作都不做, 如果生产者运行耗时较久,消费者就一直等待,如果消费者利用这个时间,那么可以处理更多请求,业务整体效率会提升

实际情况下,Dubbo有些必要的返回值必须等待,但是不必要等待的服务返回值,我们可以不等待去做别的事情, 这种情况下我们就要使用消息队列

什么是消息队列

消息队列(Message Queue)简称MQ,也称:"消息中间件", 消息队列是采用"异步"的方式来传递数据完成业务操作流程的业务处理方式(要求两个微服务项目并不需要同时完成请求)

消息队列的特征

利用异步的特性, 提高服务器的运行效率, 减少因为远程调用出现的线程等待\阻塞时间
削峰填谷:在并发峰值超过当前系统处理能力时,我们将没处理的信息保存在消息队列中,在后面出现的较闲的时间中去处理,直到所有数据依次处理完成,能够防止在并发峰值时短时间大量请求而导致的系统不稳定
消息队列的延时:因为是异步执行,请求的发起者并不知道消息何时能处理完,如果业务不能接受这种延迟,就不要使用消息队列

常见消息队列软件

Kafka:性能好\功能弱:适合大数据量,高并发的情况,大数据领域使用较多
RabbitMQ:功能强\性能一般:适合发送业务需求复杂的消息队列,java业务中使用较多
RocketMQ:阿里的
ActiveMQ:前几年流行的,老项目可能用到

消息队列的事务处理

当接收消息队列中信息的模块运行发送异常时,怎么完成事务的回滚?

在处理消息队列异常时,经常会设置一个"死信队列",将无法处理的异常信息发送到这个队列中, 死信队列没有任何处理者,通常情况下会有专人周期性的处理死信队列的消息

什么是Kafka

Kafka是由Apache软件基金会开发的一个开源流处理平台，由Scala和Java编写。该项目的目标是为处理实时数据提供一个统一、高吞吐、低延迟的平台。Kafka最初是由LinkedIn开发，并随后于2011年初开源。

Kafka Cluster(Kafka集群)

Producer:消息的发送方,也就是消息的来源,Kafka中的生产者

order就是消息的发送方,在Dubbo中order是消费者,这个身份变化了

Consumer:消息的接收方,也是消息的目标,Kafka中的消费者

stock就是消息的接收方,在Dubbo中stock是生产者,这个身份变化了

Topic:话题或主题的意思,消息的收发双方要依据同一个话题名称,才不会将信息错发给别人

Record:消息记录,就是生产者和消费者传递的信息内容,保存在指定的Topic中

Kafka的特征与优势

Kafka作为消息队列,它和其他同类产品相比,突出的特点就是性能强大

Kafka将消息队列中的信息保存在硬盘中

Kafka对硬盘的读取规则进行优化后,效率能够接近内存

硬盘的优化规则主要依靠"顺序读写,零拷贝,日志压缩等技术"

Kafka处理队列中数据的默认设置:

Kafka队列信息能够一直向硬盘中保存(理论上没有大小限制)
Kafka默认队列中的信息保存7天,可以配置这个时间,缩短这个时间可以减少Kafka的磁盘消耗

Kafka的安装和配置

必须将我们kafka软件的解压位置设置在一个根目录,文件夹名称尽量短(例如:kafka), 然后路径不要有空格和中文, 我们要创建一个空目录用于保存Kafka运行过程中产生的数据,本次创建名称为data的空目录,下面进行Kafka启动前的配置,先到D:\kafka\config下配置有文件zookeeper.properties,找到dataDir属性修改如下

dataDir=D:/data

修改完毕之后要Ctrl+S进行保存,否则修改无效!!!!

注意D盘和data文件夹名称,匹配自己电脑的真实路径和文件夹名称

还要修改server.properties配置文件

log.dirs=D:/data

启动kafka

要想启动Kafka必须先启动Zookeeper

Zookeeper介绍

Linux服务器中安装的各种软件,很多都是有动物形象的

如果这些软件在Linux中需要修改配置信息的话,就需要进入这个软件,去修改配置,每个软件都需要单独修改配置的话,工作量很大

我们使用Zookeeper之后,可以创建一个新的管理各种软件配置的文件管理系统

Linux系统中各个软件的配置文件集中到Zookeeper中

实现在Zookeeper中,可以修改服务器系统中的各个软件配置信息

长此以往,很多软件就删除了自己写配置文件的功能,而直接从Zookeeper中获取

Kafka就是需要将配置编写在Zookeeper中的软件之一

所以要先启动zookeeper才能启动kafka

Zookeeper启动

进入路径D:\kafka\bin\windows

输入cmd进入dos命令行

D:\kafka\bin\windows>kafka-server-start.bat ..\..\config\server.properties

kafka启动

D:\kafka\bin\windows>kafka-server-start.bat ..\..\config\server.properties

Kafka使用演示

启动的zookeeper和kafka的窗口不要关闭, 我们在csmall项目中编写一个kafka使用的演示, csmall-cart-webapi模块, 添加依赖

<!-- Google JSON API -->
<!-- 它是java对象和json格式字符串相互转换的工具类 -->
<dependency><groupId>com.google.code.gson</groupId><artifactId>gson</artifactId>
</dependency>
<!-- Kafka API -->
<!--  Spring整合支持Kafka的依赖  -->
<dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId>
</dependency>

修改yml文件进行配置

spring:kafka:# 定义kafka的位置bootstrap-servers: localhost:9092# consumer.group-id是一个必须配置的设置,不配置的话启动时会报错# 意思是"话题分组",这配置的目的是为了区分不同项目的话题名称# 本质上,这个分组名称会在消息发送时,自动前缀在话题名称前# 例如当前项目发送了一个话题名称为message的消息,实际传输的话题名称为csmall.messageconsumer:group-id: csmall

在SpringBoot启动类中添加启动Kafka的注解

@SpringBootApplication
@EnableDubbo
// 启动对kafka的支持
@EnableKafka
// 为了测试kafka收发消息的功能
// 我们利用SpringBoot自带的任务调用工具,周期性的向kafka发送消息
// 明确下面的注解和kafka没有必然的支持关系
@EnableScheduling
public class CsmallCartWebapiApplication {public static void main(String[] args) {SpringApplication.run(CsmallCartWebapiApplication.class, args);}}

下面我们就可以实现周期性的向kafka发送消息并接收的操作了

编写消息的发送

cart-webapi包下创建kafka包

包中创建Producer类来发送消息

生产者

// 这个类中要编写代码进行周期运行,所以要交由Spring管理
@Component
public class Producer {// 直接从Spring容器中获取能够操作Kafka的对象// 这个对象是在添加好依赖和yml配合后,启动SpringBoot时自动添加到Spring容器的// KafkaTemplate<[话题名称的类型],[消息的类型]>@Autowiredprivate KafkaTemplate<String,String> kafkaTemplate;int i=1;// 实现每隔10秒钟(10000毫秒)运行一次的方法@Scheduled(fixedRate = 10000)public void sendMessage(){// 实例化Cart对象并赋值,尝试发送给KafkaCart cart=new Cart();cart.setId(i++);cart.setCommodityCode("PC100");cart.setUserId("UU100");cart.setPrice(RandomUtils.nextInt(90)+10);cart.setCount(RandomUtils.nextInt(10)+1);//  {"id":"1","userId":"UU100",.....}// 利用gson工具,将cart对象转换为json格式字符串方便发送Gson gson=new Gson();String json=gson.toJson(cart);System.out.println("要发送的消息为:"+json);// 执行发送kafkaTemplate.send("myCart",json);}}

kafka包中创建一个叫Consumer的类来接收消息, 接收消息的类可以是本模块的类,也可以是其它模块的类,编写的代码是完全一致

// 要接收kafka的消息需要讲接收消息的对象保存到Spring容器中
// 因为KafkaTemplate是spring在管理的
@Component
public class Consumer {// SpringKafka接收消息依靠了框架提供的"监听机制"// 框架中有一个线程,一直实时关注kafka的消息情况// 如果我们指定的话题名称(myCart)接收了消息,那么这条线程就会自动调用下面的方法@KafkaListener(topics = "myCart")// 下面定义的方法就是接收到消息后运行的方法// 这个方法有参数,参数类型是固定的,参数的值就是监听器接收到的消息内容public void received(ConsumerRecord<String,String> record){// 参数类型必须是ConsumerRecord// 泛型<[话题名称的类型],[消息的类型]>// 我们可以将record视为从kafka中接收到的消息对象String json=record.value();// json可能的值: {"id":2,"commodityCode":"PC100","price":74,"count":8,"userId":"UU100"}Gson gson=new Gson();// gson也可以将json字符串转换为java对象Cart cart=gson.fromJson(json,Cart.class);System.out.println("接收到了消息:"+cart);}}