RabbitMQ消息中间件极速入门与实战

第一章RabbitMQ起步1-1 课程导航课程导航RabbitMQ简介及AMQP协议RabbitMQ安装与使用RabbitMQ核心概念与SpringBoot整合保障100%的消息可靠性投递方案落地实现1-2 RabbitMQ简介初识RabbitMQRabbitMQ是一个开源的消息代理和队列服务器用来通过普通协议在完全不同的应用之间共享数据RabbitMQ是使用Erlang语言来编写的并且RabbitMQ是基于AMQP协议的RabbitMQ简介目前很多互联网大厂都在使用RabbitMQRabbitMQ底层采用Erlang语言进行编写开源、性能优秀稳定性保障与SpringAMQP完美的整合、API丰富集群模式丰富表达式配置HA模式镜像队列模型保证数据不丢失的前提做到高可靠性、可用性AMQP全称Advanced Message Queuing ProtocolAMQP翻译高级消息队列协议AMQP协议模型1-3 RabbitMQ安装学习笔记0.安装准备 官网地址http://www.rabbitmq.com/ 安装Linux必要依赖包Linux7 下载RabbitMQ安装包 进行安装修改相关配置文件 vim /etc/hostname vim /etc/hosts 1.安装Erlang wget https://packages.erlang-solutions.com/erlang-solutions_1.0_all.deb sudo dpkg -i erlang-solutions_1.0_all.deb sudo apt-get install erlang erlang-nox 2.安装RabbitMQ echo deb http://www.rabbitmq.com/debian/ testing main | sudo tee /etc/apt/sources.list.d/rabbitmq.list wget -O- https://www.rabbitmq.com/rabbitmq-release-signing-key.asc | sudo apt-key add - sudo apt-get install rabbitmq-server 3.安装RabbitMQ web管理插件 sudo rabbitmq-plugins enable rabbitmq_management sudo systemctl restart rabbitmq-server 访问http://localhost:15672 默认用户名密码guest/guest 4.修改RabbitMQ配置 vim /usr/lib/rabbitmq/lib/rabbitmq_server-3.5.7/ebin/rabbit.app 比如修改密码、配置等等例如loopback_users中的guest只保留guest 服务启动rabbitmq-server start 服务停止rabbitmqctl app_stop1-4 RabbitMQ概念RabbitMQ的整体架构RabbitMQ核心概念Server又称Broker接受客户端的连接实现AMQP实体服务Connection连接应用程序与Broker的网络连接Channel网络信道几乎所有的操作都在Channel中进行Channel是进行消息读写的通道客户端可建立多个Channel每个Channel代表一个会话任务Message消息服务器和应用程序之间传送的数据由Properties和Body组成Properties可以对消息进行修饰比如消息的优先级、延迟等高级特性Body则就是消息体内容Virtual host虚拟机用于进行逻辑隔离最上层的消息路由一个Virtual host里面可以有若干个Exchange和Queue同一个Virtual host里面不能有相同名称的Exchange或QueueExchange交换机接收消息根据路由键转发消息到绑定的队列BindingExchange和Queue之间的虚拟连接binding中可以包含routing keyRouting key一个路由规则虚拟机可用它来确定如何路由一个特定消息Queue也称为Message Queue消息队列保存消息并将它们转发给消费者RabbitMQ消息的流转过程第二章RabbitMQ使用2-1 发送消息SpringBoot与RabbitMQ集成引入相关依赖对application.properties进行配置创建名为rabbitmq-producer的maven工程pom如下?xml version1.0 encodingUTF-8? project xmlnshttp://maven.apache.org/POM/4.0.0 xmlns:xsihttp://www.w3.org/2001/XMLSchema-instance xsi:schemaLocationhttp://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd parent artifactId47-rabbitmq/artifactId groupIdcom.myimooc/groupId version1.0-SNAPSHOT/version /parent modelVersion4.0.0/modelVersion artifactIdrabbitmq-producer/artifactId properties spring.boot.version2.0.4.RELEASE/spring.boot.version /properties dependencyManagement dependencies dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-parent/artifactId version${spring.boot.version}/version typepom/type scopeimport/scope /dependency /dependencies /dependencyManagement dependencies dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter/artifactId /dependency !--RabbitMQ依赖-- dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-amqp/artifactId /dependency !--工具类依赖-- dependency groupIdorg.apache.commons/groupId artifactIdcommons-lang3/artifactId /dependency dependency groupIdcommons-io/groupId artifactIdcommons-io/artifactId version2.5/version /dependency dependency groupIdcom.alibaba/groupId artifactIdfastjson/artifactId version1.2.36/version /dependency dependency groupIdjavax.servlet/groupId artifactIdjavax.servlet-api/artifactId scopeprovided/scope /dependency dependency groupIdorg.slf4j/groupId artifactIdslf4j-api/artifactId /dependency dependency groupIdlog4j/groupId artifactIdlog4j/artifactId version1.2.17/version /dependency dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-test/artifactId scopetest/scope /dependency /dependencies build plugins plugin groupIdorg.springframework.boot/groupId artifactIdspring-boot-maven-plugin/artifactId /plugin /plugins /build /project1.编写Order类package com.myimooc.rabbitmq.entity; import java.io.Serializable; /** * br * 标题: 订单实体br * 描述: 订单实体br * 时间: 2018/09/06br * * author zc */ public class Order implements Serializable{ private static final long serialVersionUID 6771608755338249746L; private String id; private String name; /** * 存储消息发送的唯一标识 */ private String messageId; public Order() { } public Order(String id, String name, String messageId) { this.id id; this.name name; this.messageId messageId; } Override public String toString() { return Order{ id id , name name , messageId messageId }; } public String getId() { return id; } public void setId(String id) { this.id id; } public String getName() { return name; } public void setName(String name) { this.name name; } public String getMessageId() { return messageId; } public void setMessageId(String messageId) { this.messageId messageId; } }2.编写OrderSender类package com.myimooc.rabbitmq.producer.producer; import com.myimooc.rabbitmq.entity.Order; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.support.CorrelationData; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; /** * br * 标题: 订单消息发送者br * 描述: 订单消息发送者br * 时间: 2018/09/06br * * author zc */ Component public class OrderSender { private RabbitTemplate rabbitTemplate; Autowired public OrderSender( RabbitTemplate rabbitTemplate) { this.rabbitTemplate rabbitTemplate; } /** * 发送订单 * * param order 订单 * throws Exception 异常 */ public void send(Order order) throws Exception { CorrelationData correlationData new CorrelationData(); correlationData.setId(order.getMessageId()); // exchange交换机 // routingKey路由键 // message消息体内容 // correlationData消息唯一ID this.rabbitTemplate.convertAndSend(order-exchange, order.a, order, correlationData); } }3.编写application.properties类# RabbitMQ配置 spring.rabbitmq.addresses192.168.0.105:5672 spring.rabbitmq.usernameguest spring.rabbitmq.passwordguest spring.rabbitmq.virtual-host/ spring.rabbitmq.connection-timeout15000 # Server配置 server.servlet.context-path/ server.port8080 spring.http.encoding.charsetUTF-8 spring.jackson.date-formatyyyy-MM-dd HH:mm:ss spring.jackson.time-zoneGMT8 spring.jackson.default-property-inclusionNON_NULL4.编写Application类package com.myimooc.rabbitmq.producer; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; /** * br * 标题: 启动类br * 描述: 启动类br * 时间: 2018/09/06br * * author zc */ SpringBootApplication public class Application { public static void main(String[] args) { SpringApplication.run(Application.class, args); } }5.编写OrderSenderTest类package com.myimooc.rabbitmq.producer.producer; import com.myimooc.rabbitmq.entity.Order; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.test.context.junit4.SpringRunner; import java.util.UUID; /** * br * 标题: 订单消息发送者测试br * 描述: 订单消息发送者测试br * 时间: 2018/09/06br * * author zc */ RunWith(SpringRunner.class) SpringBootTest public class OrderSenderTest { Autowired private OrderSender orderSender; Test public void testSend1() throws Exception { Order order new Order(); order.setId(201809062009010001); order.setName(测试订单1); order.setMessageId(System.currentTimeMillis() $ UUID.randomUUID().toString().replaceAll(-,)); this.orderSender.send(order); } }2-2 处理消息创建名为rabbitmq-consumer的maven工程pom如下?xml version1.0 encodingUTF-8? project xmlnshttp://maven.apache.org/POM/4.0.0 xmlns:xsihttp://www.w3.org/2001/XMLSchema-instance xsi:schemaLocationhttp://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd parent artifactId47-rabbitmq/artifactId groupIdcom.myimooc/groupId version1.0-SNAPSHOT/version /parent modelVersion4.0.0/modelVersion artifactIdrabbitmq-consumer/artifactId properties spring.boot.version2.0.4.RELEASE/spring.boot.version /properties dependencyManagement dependencies dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-parent/artifactId version${spring.boot.version}/version typepom/type scopeimport/scope /dependency /dependencies /dependencyManagement dependencies dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter/artifactId /dependency !--RabbitMQ依赖-- dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-amqp/artifactId /dependency !--工具类依赖-- dependency groupIdorg.apache.commons/groupId artifactIdcommons-lang3/artifactId /dependency dependency groupIdcommons-io/groupId artifactIdcommons-io/artifactId version2.5/version /dependency dependency groupIdcom.alibaba/groupId artifactIdfastjson/artifactId version1.2.36/version /dependency dependency groupIdjavax.servlet/groupId artifactIdjavax.servlet-api/artifactId scopeprovided/scope /dependency dependency groupIdorg.slf4j/groupId artifactIdslf4j-api/artifactId /dependency dependency groupIdlog4j/groupId artifactIdlog4j/artifactId version1.2.17/version /dependency dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-test/artifactId scopetest/scope /dependency /dependencies build plugins plugin groupIdorg.springframework.boot/groupId artifactIdspring-boot-maven-plugin/artifactId /plugin /plugins /build /project1.编写Order类package com.myimooc.rabbitmq.entity; import java.io.Serializable; /** * br * 标题: 订单实体br * 描述: 订单实体br * 时间: 2018/09/06br * * author zc */ public class Order implements Serializable{ private static final long serialVersionUID 6771608755338249746L; private String id; private String name; /** * 存储消息发送的唯一标识 */ private String messageId; public Order() { } public Order(String id, String name, String messageId) { this.id id; this.name name; this.messageId messageId; } Override public String toString() { return Order{ id id , name name , messageId messageId }; } public String getId() { return id; } public void setId(String id) { this.id id; } public String getName() { return name; } public void setName(String name) { this.name name; } public String getMessageId() { return messageId; } public void setMessageId(String messageId) { this.messageId messageId; } }2.编写OrderReceiver类package com.myimooc.rabbitmq.consumer.consumer; import com.rabbitmq.client.Channel; import com.myimooc.rabbitmq.entity.Order; import org.springframework.amqp.rabbit.annotation.*; import org.springframework.amqp.support.AmqpHeaders; import org.springframework.messaging.handler.annotation.Headers; import org.springframework.messaging.handler.annotation.Payload; import org.springframework.stereotype.Component; import java.util.Map; /** * br * 标题: 订单接收者br * 描述: 订单接收者br * 时间: 2018/09/06br * * author zc */ Component public class OrderReceiver { /** * 接收消息 * * param order 消息体内容 * param headers 消息头内容 * param channel 网络信道 * throws Exception 异常 */ RabbitListener(bindings QueueBinding( value Queue(value order-queue,durable true), exchange Exchange(name order-exchange,type topic), key order.* )) RabbitHandler public void onOrderMessage(Payload Order order, Headers MapString, Object headers, Channel channel) throws Exception { // 消费者操作 System.out.println(收到消息); System.out.println(订单信息 order.toString()); // 手动签收消息 Long deliveryTag (Long) headers.get(AmqpHeaders.DELIVERY_TAG); channel.basicAck(deliveryTag, false); } }3.编写application.properties类# RabbitMQ连接配置 spring.rabbitmq.addresses192.168.0.105:5672 spring.rabbitmq.usernameguest spring.rabbitmq.passwordguest spring.rabbitmq.virtual-host/ spring.rabbitmq.connection-timeout15000 # RabbitMQ消费配置 # 基本并发5 spring.rabbitmq.listener.simple.concurrency5 # 最大并发10 spring.rabbitmq.listener.simple.max-concurrency10 # 签收模式手动签收 spring.rabbitmq.listener.simple.acknowledge-modemanual # 限流策略同一时间只有1条消息发送过来消费 spring.rabbitmq.listener.simple.prefetch1 # Server配置 server.servlet.context-path/ server.port8082 spring.http.encoding.charsetUTF-8 spring.jackson.date-formatyyyy-MM-dd HH:mm:ss spring.jackson.time-zoneGMT8 spring.jackson.default-property-inclusionNON_NULL4.编写Application类package com.myimooc.rabbitmq.consumer; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; /** * br * 标题: 启动类br * 描述: 启动类br * 时间: 2018/09/06br * * author zc */ SpringBootApplication public class Application { public static void main(String[] args) { SpringApplication.run(Application.class, args); } }第三章可靠性投递3-1 设计方案保障100%消息投递成功设计方案一3-2 代码详解因篇幅限制源码请到github地址查看这里仅展示核心关键类1.编写OrderSender类package com.myimooc.rabbitmq.ha.producer; import com.myimooc.rabbitmq.entity.Order; import com.myimooc.rabbitmq.ha.constant.Constants; import com.myimooc.rabbitmq.ha.dao.mapper.BrokerMessageLogMapper; import com.myimooc.rabbitmq.ha.dao.po.BrokerMessageLogPO; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.support.CorrelationData; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; /** * br * 标题: 订单消息发送者br * 描述: 订单消息发送者br * 时间: 2018/09/06br * * author zc */ Component public class OrderSender { Autowired private RabbitTemplate rabbitTemplate; Autowired private BrokerMessageLogMapper brokerMessageLogMapper; /** * 回调方法confirm确认 */ private final RabbitTemplate.ConfirmCallback confirmCallback new RabbitTemplate.ConfirmCallback() { Override public void confirm(CorrelationData correlationData, boolean ack, String cause) { System.out.println(correlationData correlationData); String messageId correlationData.getId(); if (ack) { // 如果confirm返回成功则进行更新 BrokerMessageLogPO messageLogPO new BrokerMessageLogPO(); messageLogPO.setMessageId(messageId); messageLogPO.setStatus(Constants.OrderSendStatus.SEND_SUCCESS); brokerMessageLogMapper.changeBrokerMessageLogStatus(messageLogPO); } else { // 失败则进行具体的后续操作重试或者补偿等 System.out.println(异常处理...); } } }; /** * 发送订单 * * param order 订单 */ public void send(Order order) { // 设置回调方法 this.rabbitTemplate.setConfirmCallback(confirmCallback); // 消息ID CorrelationData correlationData new CorrelationData(order.getMessageId()); // 发送消息 this.rabbitTemplate.convertAndSend(order-exchange, order.a, order, correlationData); } }2.编写OrderService类package com.myimooc.rabbitmq.ha.service; import com.myimooc.rabbitmq.entity.Order; import com.myimooc.rabbitmq.ha.constant.Constants; import com.myimooc.rabbitmq.ha.dao.mapper.BrokerMessageLogMapper; import com.myimooc.rabbitmq.ha.dao.mapper.OrderMapper; import com.myimooc.rabbitmq.ha.dao.po.BrokerMessageLogPO; import com.myimooc.rabbitmq.ha.producer.OrderSender; import com.myimooc.rabbitmq.ha.util.FastJsonConvertUtils; import org.apache.commons.lang3.time.DateUtils; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; import java.util.Date; /** * br * 标题: 订单服务br * 描述: 订单服务br * 时间: 2018/09/07br * * author zc */ Service public class OrderService { Autowired private OrderMapper orderMapper; Autowired private BrokerMessageLogMapper brokerMessageLogMapper; Autowired private OrderSender orderSender; /** * 创建订单 * * param order 订单 */ public void create(Order order) { // 当前时间 Date orderTime new Date(); // 业务数据入库 this.orderMapper.insert(order); // 消息日志入库 BrokerMessageLogPO messageLogPO new BrokerMessageLogPO(); messageLogPO.setMessageId(order.getMessageId()); messageLogPO.setMessage(FastJsonConvertUtils.convertObjectToJson(order)); messageLogPO.setTryCount(0); messageLogPO.setStatus(Constants.OrderSendStatus.SENDING); messageLogPO.setNextRetry(DateUtils.addMinutes(orderTime, Constants.ORDER_TIMEOUT)); this.brokerMessageLogMapper.insert(messageLogPO); // 发送消息 this.orderSender.send(order); } }3.编写RetryMessageTask类package com.myimooc.rabbitmq.ha.task; import com.myimooc.rabbitmq.entity.Order; import com.myimooc.rabbitmq.ha.constant.Constants; import com.myimooc.rabbitmq.ha.dao.mapper.BrokerMessageLogMapper; import com.myimooc.rabbitmq.ha.dao.po.BrokerMessageLogPO; import com.myimooc.rabbitmq.ha.producer.OrderSender; import com.myimooc.rabbitmq.ha.util.FastJsonConvertUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Component; import java.util.List; /** * br * 标题: 重发消息定时任务br * 描述: 重发消息定时任务br * 时间: 2018/09/07br * * author zc */ Component public class RetryMessageTask { private Logger logger LoggerFactory.getLogger(getClass()); Autowired private OrderSender orderSender; Autowired private BrokerMessageLogMapper brokerMessageLogMapper; /** * 启动完成3秒后开始执行每隔10秒执行一次 */ Scheduled(initialDelay 3000, fixedDelay 10000) public void retrySend() { logger.debug(重发消息定时任务开始); // 查询 status 0 和 timeout 的消息日志 ListBrokerMessageLogPO pos this.brokerMessageLogMapper.listSendFailureAndTimeoutMessage(); for (BrokerMessageLogPO po : pos) { logger.debug(处理消息日志{},po); if (po.getTryCount() Constants.MAX_RETRY_COUNT) { // 更新状态为失败 BrokerMessageLogPO messageLogPO new BrokerMessageLogPO(); messageLogPO.setMessageId(po.getMessageId()); messageLogPO.setStatus(Constants.OrderSendStatus.SEND_FAILURE); this.brokerMessageLogMapper.changeBrokerMessageLogStatus(messageLogPO); } else { // 进行重试重试次数1 this.brokerMessageLogMapper.updateRetryCount(po); Order reSendOrder FastJsonConvertUtils.convertJsonToObject(po.getMessage(), Order.class); try { this.orderSender.send(reSendOrder); } catch (Exception ex) { // 异常处理 logger.error(消息发送异常{}, ex); } } } logger.debug(重发消息定时任务结束); } }4.编写ApplicationTest类package com.myimooc.rabbitmq.ha; import com.myimooc.rabbitmq.entity.Order; import com.myimooc.rabbitmq.ha.service.OrderService; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.test.context.junit4.SpringRunner; import java.util.UUID; /** * br * 标题: 订单创建测试br * 描述: 订单创建测试br * 时间: 2018/09/07br * * author zc */ RunWith(SpringRunner.class) SpringBootTest public class ApplicationTest { Autowired private OrderService orderService; Test public void testCreateOrder(){ Order order new Order(); order.setId(String.valueOf(System.currentTimeMillis())); order.setName(测试创建订单); order.setMessageId(System.currentTimeMillis() $ UUID.randomUUID().toString().replaceAll(-,)); this.orderService.create(order); } }