目录
消息队列
MQ作⽤
1 处理消息规则
JMS
JMS消息模型
JMS消息种类
AMQP
AMQP消息种类：
AMQP消息模型
MQTT
2.Spring整合ActiveMQ
安装（以windows为例）
解压即可
启动服务器
访问web管理服务
整合
1.导⼊springboot整合ActiveMQ的starter
2.配置ActiveMQ的服务器地址
3.使⽤JmsMessagingTemplate操作ActiveMQ
4.使⽤消息监听器
5.切换消息模型
3. SpringBoot整合RabbitMQ
Erlang安装（windows版）
安装RabbitMQ
启动服务器
访问web管理服务
整合(direct模型)
1.导⼊springboot整合amqp的starter
2.配置RabbitMQ的服务器地址
3.初始化直连模式系统设置
4.使⽤AmqpTemplate操作RabbitMQ
5.使⽤消息监听器
整合(topic模型)
1.初始化主题模式系统设置
2.使⽤AmqpTemplate操作RabbitMQ
3.使⽤消息监听器在服务器启动后，监听指定队列
4 SpringBoot整合RocketMQ
下载:
RocketMQ⼯作模式
启动服务器
测试服务器启动状态
整合
1.导⼊springboot整合RocketMQ的starter
2.配置RocketMQ的服务器地址
3.使⽤RocketMQTemplate操作RocketMQ
4.使⽤消息监听器
5.SpringBoot整合Kafka
下载安装
启动服务器
创建主题
测试服务器启动状态
整合
1.导⼊springboot整合Kafka的starter
2.配置Kafka的服务器地址
3.使⽤KafkaTemplate操作Kafka
4.使⽤消息监听器
消息队列
MQ（Message Queue）消息队列，是基础数据结构中“先进先出”的⼀种数据结构。指把要传输的数据（消息）放在队列中，⽤队列机制来实现消息传递——⽣产者产⽣消息并把消息放⼊队列，然后由消费者去处理。消费者可以到指定队列拉取消息，或者订阅相应的队列，由MQ服务端给其推送消息。
MQ作⽤
1.流量消峰
在⼤量请求时(秒杀场景)，使⽤消息队列做缓冲处理，削弱峰值流量，防⽌系统在短时间内被峰值流量冲垮。
2.应⽤解耦
使⽤消息MQ后，只需要保证消息格式不变，不需要关⼼发布者及消费者之间的关系，这两者不需要彼此联系
3.异步处理
对于消息的⽣产者与消费者的⼯作模式，还可以将消息划分成两种模式，同步消费与异步消息。
1 处理消息规则
⽬前企业级开发中⼴泛使⽤的消息处理技术共三⼤类，具体如下：
JMSAMQPMQTT
这些都是规范，就想JDBC技术，是个规范，开发针对规范开发，运⾏还要靠实现类，例如MySQL提供了JDBC的实现，最终运⾏靠的还是实现。并且这三类规范都是针对异步消息进⾏处理的，也符合消息的设计本质，处理异步的业务。
JMS
JMS（Java Message Service）,这是⼀个规范，作⽤等同于JDBC规范，提供了与消息服务相关的API接⼝。
JMS消息模型
JMS规范中规范了消息有两种模型。分别是点对点模型和发布订阅模型。
点对点模型：peer-2-peer，⽣产者会将消息发送到⼀个保存消息的容器中，通常使⽤队列模型，使⽤队列保存消息。⼀个队列的消息只能被⼀个消费者消费，或未被及时消费导致超时。这种模型下，⽣产者和消费者是⼀对⼀绑定的。
发布订阅模型：publish-subscribe，⽣产者将消息发送到⼀个保存消息的容器中，也是使⽤队列模型来保存。但是消息可以被多个消费者消费，⽣产者和消费者完全独⽴，相互不需要感知对⽅的存在。
以上这种分类是从消息的⽣产和消费过程来进⾏区分，针对消息所包含的信息不同，还可以进⾏不同类别的划分。
JMS消息种类
根据消息中包含的数据种类划分，可以将消息划分成6种消息。
TextMessageMapMessageBytesMessageStreamMessageObjectMessageMessage （只有消息头和属性）
JMS主张不同种类的消息，消费⽅式不同，可以根据使⽤需要选择不同种类的消息。例如ActiveMQ、Redis、HornetMQ。但是也有⼀些不太规范的实现，参考JMS的标准设计，但是⼜不完全满⾜其规范，例如：RabbitMQ、RocketMQ。
AMQP
JMS的设计是J2EE规范，站在Java开发的⻆度思考问题。但是现实往往是复杂度很⾼的。⽐如我有⼀个.NET开发的系统A，有⼀个Java开发的系统B，现在要从A系统给B系统发业务消息，结果两边数据格式不统⼀，没法操作。JMS不是可以统⼀数据格式吗？提供了6种数据种类，因为A系统的底层语⾔不是Java语⾔开发的，根本不⽀持那些对象。这就意味着如果想使⽤现有的业务系统A继续开发已经不可能了，必须推翻重新做使⽤Java语⾔开发的A系统。
AMQP的出现解决的是消息传递时使⽤的消息种类的问题，化繁为简，但是其并没有完全推翻JMS的操作API，所以说AMQP仅仅是⼀种协议，规范了数据传输的格式⽽已。
AMQP（advanced message queuing protocol）：⼀种协议（⾼级消息队列协议，也是消息代理规范），规范了⽹络交换的数据格式，兼容JMS操作。
优点 具有跨平台性，服务器供应商，⽣产者，消费者可以使⽤不同的语⾔来实现
AMQP消息种类：
byte[]
AMQP在JMS的消息模型基础上⼜进⾏了进⼀步的扩展，除了点对点和发布订阅的模型，开发了⼏种全新的消息模型，适应各种各样的消息发送。
AMQP消息模型
direct exchangefanout exchangetopic exchangeheaders exchangesystem exchange
⽬前实现了AMQP协议的消息中间件技术也很多，⽽且都是较为流⾏的技术，例如：RabbitMQ、StormMQ、RocketMQ
MQTT
MQTT（Message Queueing Telemetry Transport）消息队列遥测传输，专为⼩设备设计，是物联⽹（IOT）⽣态系统中主要成分之⼀。由于与JavaEE企业级开发没有交集，此处不作过多的说明。
2.Spring整合ActiveMQ
ActiveMQ是MQ产品中的元⽼级产品，早期标准MQ产品之⼀，在AMQP协议没有出现之前，占据了消息中间件市场的绝⼤部分份额，后期因为AMQP系列产品的出现，迅速⾛弱，⽬前仅在⼀些线上运⾏的产品中出现，新产品开发较少采⽤。
安装（以windows为例）
官方下载地址https://activemq.apache.org/components/classic/download/
解压即可
启动服务器
运⾏bin⽬录下的win32或win64⽬录下的activemq.bat命令即可，根据⾃⼰的操作系统选择即可，默认对外服务端⼝61616。
访问web管理服务
ActiveMQ启动后会启动⼀个Web控制台服务，可以通过该服务管理ActiveMQ。
http://127.0.0.1:8161/
web管理服务默认端⼝8161，访问后可以打开ActiveMQ的管理界⾯，如下：
⾸先输⼊访问⽤户名和密码，初始化⽤户名和密码相同，均为：admin，成功登录
点击Mange ActiveMQ broker进入主页面
进⼊管理后台界⾯，如下：
看到上述界⾯视为启动ActiveMQ服务成功。
点对点消息列表
列表各列信息含义如下：
Number Of Pending Messages ：等待消费的消息 这个是当前未出队列的数量。
Number Of Consumers ：消费者 这个是消费者端的消费者数量
Messages Enqueued ：进入队列的消息 进入队列的总数量,包括出队列的。
Messages Dequeued ：出了队列的消息 可以理解为是消费这消费掉的数量。
整合
需求: 订单业务发送短信需求。
处理订单信息的时候会有消息发送。因此会有消息的创建和消息的消费。 我们模拟消息的创建在队列中，消费的时候从队列中获取消息即可。
1.导⼊springboot整合ActiveMQ的starter
2.配置ActiveMQ的服务器地址
server:
port: 80
spring:
activemq:
# 地址端口
broker-url: tcp://localhost:61616
jms:
template:
# 给mq起一个名字，不然报错
default-destination: zzx
3.使⽤JmsMessagingTemplate操作ActiveMQ
创建消息服务
public interface MessageService {
String doMessage();
void sendMessage(String id);
}
@Service
public class MessageServiceImpl implements MessageService {
@Autowired
private JmsMessagingTemplate messagingTemplate;
//消费消息，从mq中取消息消费
@Override
public String doMessage() {
String id = messagingTemplate.receiveAndConvert(String.class);
System.out.println("已完成信息业务 id=" + id);
return id;
}
//创建消息，并发送给订单（生产者）
@Override
public void sendMessage(String id) {
System.out.println("发送的消息已经进入mq队列 id=" + id);
messagingTemplate.convertAndSend(id);
}
}
发送消息需要先将消息的类型转换成字符串，然后再发送，所以是convertAndSend，定义消息发送的位置，和具体的消息内容，此处使⽤id作为消息
接收消息需要先将消息接收到，然后再转换成指定的数据类型，所以是receiveAndConvert，接收消息除了提供读取的位置，还要给出转换后的数据的具体类型。 内容。
创建订单服务
/**
* 处理订单的servic
*/
public interface OrderService {
public void order(String id);
}
@Service
public class OrderServiceImpl implements OrderService {
@Autowired
private MessageService messageService;
@Override
public void order(String id) {
System.out.println("订单开始处理...");
//订单处理
messageService.sendMessage(id);
System.out.println("订单处理结束.");
}
}
创建controller
@RestController
@RequestMapping("/msgs")
public class MessageController {
@Autowired
private MessageService service;
//消费消息
@GetMapping
public String doMessage(){
String id = service.doMessage();
return id;
}
}
/**
* 订单表现层
*/
@RestController
@RequestMapping("/orders")
public class OrderController{
@Autowired
private OrderService orderService;
@PostMapping("{id}")
public void order(@PathVariable String id){
orderService.order(id);
}
}
postman发送post创建消息请求http://localhost/orders/1
postman发送get消费消息请求http://localhost/msgs
如果将消息发送到某个mq队列也可以这样设置
@Service
public class MessageServiceImpl implements MessageService {
@Autowired
private JmsMessagingTemplate messagingTemplate;
//消费消息，从mq中取消息消费
@Override
public String doMessage() {
//String id = messagingTemplate.receiveAndConvert(String.class);
String id = messagingTemplate.receiveAndConvert("order.queue.id", String.class);
System.out.println("已完成信息业务 id=" + id);
return id;
}
//创建消息，并发送给订单（生产者）
@Override
public void sendMessage(String id) {
System.out.println("发送的消息已经进入mq队列 id=" + id);
//messagingTemplate.convertAndSend(id);
messagingTemplate.convertAndSend("order.queue.id",id);
}
}
上述使用postman发送请求
4.使⽤消息监听器
在服务器启动后，监听指定位置，当消息出现后，⽴即消费消息
@Component
public class MessageListener {
//配置监听器监听队列
@JmsListener(destination = "order.queue.id")
//转发到队列
@SendTo("order.queue.demo")
public String recrive(String id){
System.out.println("已完成信息业务 id=" + id);
return "listener: "+id;
}
}
使⽤注解@JmsListener定义当前⽅法监听ActiveMQ中指定名称的消息队列。
如果当前消息队列处理完还需要继续向下传递当前消息到另⼀个队列中使⽤注解@SendTo即可，这样即可构造连续执⾏的顺序消息队列。
5.切换消息模型
由点对点模型到发布订阅模型，修改jms配置即可
server:
port: 80
spring:
activemq:
# 地址端口
broker-url: tcp://localhost:61616
jms:
template:
# 给mq起一个名字，不然报错
default-destination: zzx
#由点对点模型到发布订阅模型
pub-sub-domain: true
pub-sub-domain默认值为false，即点对点模型，修改为true后就是发布订阅模型。
总结 1. springboot整合ActiveMQ提供了JmsMessagingTemplate对象作为客户端操作消息队列 2. 操作ActiveMQ需要配置ActiveMQ服务器地址，默认端⼝61616 3. 企业开发时通常使⽤监听器来处理消息队列中的消息，设置监听器使⽤注解@JmsListener 4. 配置jms的pub-sub-domain属性可以在点对点模型和发布订阅模型间切换消息模型
3. SpringBoot整合RabbitMQ
RabbitMQ是MQ产品中的⽬前较为流⾏的产品之⼀，它遵从AMQP协议。 RabbitMQ的底层实现语⾔使⽤的是Erlang，所以安装RabbitMQ需要先安装Erlang。
RabbitMQ与Erlang版本对应表RabbitMQ Erlang Version Requirements — RabbitMQ
Erlang安装（windows版）
安装包下载地址Downloads - Erlang/OTP
下载完毕后得到exe安装⽂件，⼀键傻⽠式安装，安装完毕需要重启，需要重启，需要重启。
安装的过程中可能会出现依赖Windows组件的提示，根据提示下载安装即可，都是⾃动执⾏的，如下：
Erlang安装后需要配置环境变量，否则RabbitMQ将⽆法找到安装的Erlang。需要配置项如下，作⽤等同JDK配置环境变量的作⽤。
ERLANG_HOMEPATH
安装RabbitMQ
下载地址：Installing on Windows — RabbitMQ
下载完毕后得到exe安装⽂件，⼀键傻⽠式安装
启动服务器
以管理员身份在安装目录的sbin目录下执行命令
rabbitmq-service.bat start # 启动服务 rabbitmq-service.bat stop # 停⽌服务 rabbitmqctl status # 查看服务状态
或者运⾏sbin⽬录下的rabbitmq-service.bat ，默认对外服务端⼝5672，后台管理端口15672。
访问web管理服务
RabbitMQ也提供有web控制台服务，但是此功能是⼀个插件，需要先启⽤才可以使⽤。
rabbitmq-plugins.bat list # 查看当前所有插件的运⾏状态 rabbitmq-plugins.bat enable rabbitmq_management # 安装rabbitmq_management插件
启动插件后可以在插件运⾏状态中查看是否运⾏，运⾏后通过浏览器即可打开服务后台管理界⾯
http://localhost:15672
web管理服务默认端⼝15672，访问后可以打开RabbitMQ的管理界⾯，如下：
⾸先输⼊访问⽤户名和密码，初始化⽤户名和密码相同，均为：guest，成功登录后进⼊管理后台界⾯，如下：
整合(direct模型)
RabbitMQ满⾜AMQP协议，因此不同的消息模型对应的制作不同，先使⽤最简单的direct模型开发。
1.导⼊springboot整合amqp的starter
amqp协议默认实现为rabbitmq⽅案
2.配置RabbitMQ的服务器地址
server:
port: 80
spring:
rabbitmq:
#地址
host: localhost
#对外提供服务端口
port: 5672
3.初始化直连模式系统设置
由于RabbitMQ不同模型要使⽤不同的交换机，因此需要先初始化RabbitMQ相关 的对象，例如队列，交换机等
@Configuration
public class RabbitConfigDirect {
//创建消息队列
@Bean
public Queue directQueue(){
return new Queue("direct_queue");
}
@Bean
public Queue directQueue2(){
return new Queue("direct_queue2");
}
//创建交换机
@Bean
public DirectExchange directExchange(){
return new DirectExchange("directExchange");
}
//交换机和队列绑定
@Bean
public Binding bindingDirect(){
return BindingBuilder.bind(directQueue()).to(directExchange()).with("direct");
}
@Bean
public Binding bindingDirect2(){
return BindingBuilder.bind(directQueue2()).to(directExchange()).with("direct2");
}
}
队列Queue与直连交换机DirectExchange创建后，还需要绑定他们之间的关系Binding，这样就可以通过交换机操作对应队列。
4.使⽤AmqpTemplate操作RabbitMQ
创建服务
public interface MessageService {
String doMessage();
void sendMessage(String id);
}
@Service
public class MessageServiceRabbitMQDirectImpl implements MessageService {
@Autowired
private AmqpTemplate amqpTemplate;
@Override
public void sendMessage(String id) {
System.out.println("待发送短信的订单已纳⼊处理队列（rabbitmqdirect），id："+id);
amqpTemplate.convertAndSend("directExchange","direct",id);
}
@Override
public String doMessage() {
return null;
}
}
/**
* 处理订单的servic
*/
public interface OrderService {
public void order(String id);
}
@Service
public class OrderServiceImpl implements OrderService {
@Autowired
private MessageService messageService;
@Override
public void order(String id) {
System.out.println("订单开始处理...");
//订单处理
messageService.sendMessage(id);
System.out.println("订单处理结束.");
}
}
5.使⽤消息监听器
在服务器启动后，监听指定位置，当消息出现后，⽴即消费消息
@Component
public class RabbitMessageListener {
//配置监听器监听队列
@RabbitListener(queues = "direct_queue")
public void recrive(String id){
System.out.println("已经完成消息获取 id=" + id);
}
}
使⽤注解@RabbitListener定义当前⽅法监听RabbitMQ中指定名称的消息队列。
创建controller
/**
* 订单表现层
*/
@RestController
@RequestMapping("/orders")
public class OrderController{
@Autowired
private OrderService orderService;
@PostMapping("{id}")
public void order(@PathVariable String id){
orderService.order(id);
}
}
postman发送post请求http://localhost/orders/3
整合(topic模型)
1.初始化主题模式系统设置
@Configuration
public class RabbitConfigTopic {
//创建消息队列
@Bean
public Queue topicQueue(){
return new Queue("topic_queue");
}
//创建交换机
@Bean
public TopicExchange topicExchange(){
return new TopicExchange("topicExchange");
}
//交换机和队列绑定
@Bean
public Binding bindingTopic(){
return BindingBuilder.bind(topicQueue()).to(topicExchange()).with("topic.order.id");
}
}
主题模式⽀持routingKey匹配模式，*表示匹配⼀个单词，#表示匹配任意内容，这样就可以通过主题交换机将消息分发到不同的队列中
2.使⽤AmqpTemplate操作RabbitMQ
@Service
public class MessageServiceRabbitMQTopicImpl implements MessageService {
@Autowired
private AmqpTemplate amqpTemplate;
@Override
public void sendMessage(String id) {
System.out.println("待发送短信的订单已纳⼊处理队列（rabbitmq topic），id："+id);
amqpTemplate.convertAndSend("topicExchange","topic.order.id",id);
}
@Override
public String doMessage() {
return null;
}
}
发送消息后，根据当前提供的routingKey与绑定交换机时设定的routingKey进⾏匹配，规则匹配成功消息才会进⼊到对应的队列中。
3.使⽤消息监听器在服务器启动后，监听指定队列
@Component
public class RabbitMessageTopicListener {
//配置监听器监听队列
@RabbitListener(queues = "topic_queue")
public void recrive(String id){
System.out.println("已经完成消息获取 id=" + id);
}
}
使⽤注解@RabbitListener定义当前⽅法监听RabbitMQ中指定名称的消息队列。
其他与direct模型一样
总结
1. springboot整合RabbitMQ提供了AmqpTemplate对象作为客户端操作消息队列 2. 操作ActiveMQ需要配置ActiveMQ服务器地址，默认端⼝5672 3. 企业开发时通常使⽤监听器来处理消息队列中的消息，设置监听器使⽤注@RabbitListener 4. RabbitMQ有5种消息模型，使⽤的队列相同，但是交换机不同。交换机不同，对应的消息进⼊的策略也不同
4 SpringBoot整合RocketMQ
RocketMQ 是阿⾥巴巴在2012年开源的分布式消息中间件，⽬前已经捐赠给Apache 软件基⾦会，并于2017年9⽉25⽇成为 Apache 的顶级项⽬。作为经历过多次阿⾥巴巴双⼗⼀的洗礼并有稳定出⾊表现的国产中间件，以其⾼性能、低延时和⾼可靠等特性近年来已经也被越来越多的企业使⽤，它遵从AMQP协议。
下载:
下载地址：下载 | RocketMQ Binary 下载
RocketMQ安装后需要配置环境变量，具体如下：
ROCKETMQ_HOMEPATHNAMESRV_ADDR （建议）： 127.0.0.1:9876
RocketMQ是基于jdk8环境完成的，所以最好有一个免安装的jdk8
RocketMQ⼯作模式
在RocketMQ中，处理业务的服务器称为broker，⽣产者与消费者不是直接与broker联系的，⽽是通过命名服务器进⾏通信。broker启动后会通知命名服务器⾃⼰已经上线，这样命名服务器中就保存有所有的broker信息。当⽣产者与消费者需要连接broker时，通过命名服务器找到对应的处理业务的broker，因此命名服务器在整套结构中起到⼀个信息中⼼的作⽤。并且broker启动前必须保障命名服务器先启动。
启动服务器
在jdk8的环境下，在安装目录的bin目录下运行
mqnamesrv.cmd # 启动命名服务器 mqbroker.cmd # 启动broker
测试服务器启动状态
RocketMQ提供有⼀套测试服务器功能的测试程序，运⾏bin⽬录下的tools命令即可使⽤。
tools org.apache.rocketmq.example.quickstart.Producer # ⽣产消息 tools org.apache.rocketmq.example.quickstart.Consumer # 消费消息
整合
1.导⼊springboot整合RocketMQ的starter
此坐标不由springboot维护版本
2.配置RocketMQ的服务器地址
rocketmq:
#地址
name-server: loclhost:9876
#生产者组
producer:
group: group_rocketmq
设置默认的⽣产者消费者所属组group。
3.使⽤RocketMQTemplate操作RocketMQ
@Service
public class MessageServiceRocketMQImpl implements MessageService {
@Autowired
private RocketMQTemplate rocketMQTemplate;
@Override
public void sendMessage(String id) {
System.out.println("待发送短信的订单已纳⼊处理队列（rocketMQ），id："+id);
//rocketMQTemplate.convertAndSend("order_id",id);//同步执行
//异步执行
SendCallback callback = new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
System.out.println("消息发送成功");
}
@Override
public void onException(Throwable throwable) {
System.out.println("消息发送失败");
}
};
rocketMQTemplate.asyncSend("order_id",id,callback);
}
}
使⽤asyncSend⽅法发送异步消息。
4.使⽤消息监听器
监听指定位置，当消息出现后，⽴即消费消息
@Component
@RocketMQMessageListener(topic = "order_id",consumerGroup = "group_rocketmq")
public class MessageListener implements RocketMQListener
@Override
public void onMessage(String id) {
System.out.println("已完成信息业务 id=" + id);
}
}
RocketMQ的监听器必须按照标准格式开发，实现RocketMQListener接⼝，泛型为消息类型。
使⽤注解@RocketMQMessageListener定义当前类监听RabbitMQ中指定组、指定名称的消息队列。
总结 1. springboot整合RocketMQ使⽤RocketMQTemplate对象作为客户端操作消息队列 2. 操作RocketMQ需要配置RocketMQ服务器地址，默认端⼝9876 3. 企业开发时通常使⽤监听器来处理消息队列中的消息，设置监听器使⽤注解 @RocketMQMessageListener
5.SpringBoot整合Kafka
Kafka是由Apache软件基⾦会开发的⼀个开源流处理平台，由Scala和Java编写。Kafka 是⼀种⾼吞吐量的分布式发布订阅消息系统，它可以处理消费者在⽹站中的所有动作流数据。
下载安装
下载地址Apache Kafka
解压安装即可
启动服务器
kafka服务器的功能相当于RocketMQ中的broker，kafka运⾏还需要⼀个类似于命名服务器的服务。在kafka安装⽬录中⾃带⼀个类似于命名服务器的⼯具，叫做zookeeper，它的作⽤是注册中⼼
在安装目录的bin目录里的windows目录下执行命令
zookeeper-server-start.bat ..\..\config\zookeeper.properties # 启动zookeeper kafka-server-start.bat ..\..\config\server.properties # 启动kafka
运⾏bin⽬录下的windows⽬录下的zookeeper-server-start命令即可启动注册中⼼，默认对外服务端⼝2181。 运⾏bin⽬录下的windows⽬录下的kafka-server-start命令即可启动kafka服务器，默认对外服务端⼝9092。
创建主题
kakfa也是基于主题操作，操作之前需要先初始化topic。
# 创建topic kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test # 查询topic kafka-topics.bat --zookeeper 127.0.0.1:2181 --list # 删除topic kafka-topics.bat --delete --zookeeper localhost:2181 --topic test
测试服务器启动状态
Kafka提供有⼀套测试服务器功能的测试程序，运⾏bin⽬录下的windows⽬录下的命令即可使⽤
kafka-console-producer.bat --broker-list localhost:9092 --topic test # 测试⽣产消息 kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic test --from-beginning # 测试消息消费
整合
1.导⼊springboot整合Kafka的starter
此坐标由springboot维护版本
2.配置Kafka的服务器地址
spring:
kafka:
#地址
bootstrap-servers: localhost:9092
# id
consumer:
group-id: order
设置默认的⽣产者消费者所属组id。
3.使⽤KafkaTemplate操作Kafka
@Service
public class MessageServiceKafkaImpl implements MessageService {
@Autowired
private KafkaTemplate
@Override
public void sendMessage(String id) {
System.out.println("待发送短信的订单已纳⼊处理队列（Kafka），id："+id);
kafkaTemplate.send("test",id);
}
}
使⽤send⽅法发送消息，需要传⼊topic名称。
4.使⽤消息监听器
服务器启动后，监听指定位置，当消息出现后，⽴即消费消息
@Component
public class MessageListener{
@KafkaListener(topics = "test")
public void onMessage(ConsumerRecord
System.out.println("已完成信息业务 id=" + record.value());
}
}
使⽤注解@KafkaListener定义当前⽅法监听Kafka中指定topic的消息，接收到的消息封装在对象ConsumerRecord中，获取数据从ConsumerRecord对象中获取即可。
总结 1. springboot整合Kafka使⽤KafkaTemplate对象作为客户端操作消息队列 2. 操作Kafka需要配置Kafka服务器地址，默认端⼝9092 3. 企业开发时通常使⽤监听器来处理消息队列中的消息，设置监听器使⽤注解 @KafkaListener。接收消息保存在形参ConsumerRecord对象中
精彩内容
您阅读本篇文章共花了：
发表评论