RabbitMQ 学习笔记hi这些是我整理的rabbitmq相关的笔记具体怎么安装在这里就不一一阐述了不知道的可以问一下ai或者搜索相关视频我将rabbitmq学习分为三天进行最好对着步骤进行实操效果更佳。一、RabbitMQ 是什么RabbitMQ 是一个消息队列中间件负责在两个系统之间传递消息。打个比方你去餐厅吃饭点完单后拿到一个号码牌消息后厨按号码牌顺序做菜消费消息你不用站在厨房门口等可以先去找座位异步处理后厨忙不过来单子在排队流量削峰核心角色角色作用举例生产者Producer发消息的人用户下单系统发一条消息消费者Consumer收消息的人库存服务收到消息扣减库存队列Queue存消息的地方消息在队列里排队等待处理交换机Exchange路由消息到队列根据规则决定消息发给哪个队列二、为什么需要 RabbitMQ1. 异步处理用户下单后需要扣库存、发短信、记日志。不用 RabbitMQ下单 → 扣库存 → 发短信 → 记日志 → 返回全部串行慢用 RabbitMQ下单 → 扣库存 → 发消息到队列 → 返回快短信和日志异步处理2. 系统解耦订单系统不直接调用库存系统、短信系统而是发消息到队列各系统自己来取消息。好处订单系统不需要知道谁在用这个消息加新系统不用改订单代码。3. 流量削峰秒杀活动瞬间涌入大量请求先全部放进队列排队后端按自己能力慢慢处理不会被压垮。三、RabbitMQ 核心概念1. 消息流转过程生产者 → 交换机 → 队列 → 消费者生产者不直接把消息发给队列而是先发给交换机交换机再根据规则转发到队列。2. 交换机类型类型特点场景Direct精确匹配路由键点对点精确路由Fanout广播给所有绑定队列消息广播Topic通配符匹配路由键灵活的消息分类Headers按消息头匹配很少使用3. 路由键Routing Key消息带的一个标签交换机根据这个标签决定消息发给哪个队列。例如消息路由键是order.payDirect 交换机只转发给绑定了order.pay的队列。4. 绑定Binding队列和交换机之间的连接关系可以理解为这个队列订阅了交换机的哪类消息。四、管理界面操作启动 RabbitMQ 后打开浏览器访问http://localhost:15672账号guest密码guest界面主要功能页面功能Overview查看整体状态、连接数、消息数量Connections查看当前连接Channels查看信道一个连接可以有多个信道Exchanges查看和管理交换机Queues查看和管理队列Admin用户管理、权限管理手动创建队列点击顶部Queues标签点击Add a new queueName 填test_queue点击Add queue手动发送消息进入刚创建的队列test_queue展开Publish messagePayload 填hello rabbitmq点击Publish message手动接收消息在队列页面展开Get messages点击Get Message(s)可以看到刚才发送的消息内容五、五种消息模式1. 简单模式Simple最基础的模式一个生产者一个队列一个消费者。生产者 → 队列 → 消费者生产者往队列里发消息消费者从队列里取消息一对一一条消息只被一个消费者处理2. 工作队列模式Work Queue一个生产者一个队列多个消费者。生产者 → 队列 → 消费者1 → 消费者2 → 消费者3多个消费者竞争消费同一个队列的消息默认轮询分发每人一条平均分配。适合场景任务分发多个工作者抢活干。3. 发布订阅模式Fanout一个生产者一个 Fanout 交换机多个队列多个消费者。生产者 → Fanout交换机 → 队列1 → 消费者1 → 队列2 → 消费者2Fanout 交换机把消息广播给所有绑定的队列每个队列都能收到完整的一份消息。适合场景一条消息需要多个系统同时处理如同时通知库存系统和日志系统。4. 路由模式Direct一个生产者一个 Direct 交换机多个队列。生产者 → Direct交换机 →路由键error→ 队列1绑定了error→ 消费者1 →路由键info → 队列2绑定了info → 消费者2Direct 交换机根据消息的路由键精确匹配只转发给路由键匹配的队列。适合场景按消息类型分流错误日志和普通日志分开处理。5. 主题模式Topic一个生产者一个 Topic 交换机多个队列。生产者 → Topic交换机 →路由键order.pay→ 队列1绑定了order.*→ 消费者1 → 队列2绑定了order.#→ 消费者2Topic 交换机支持通配符匹配通配符含义示例*匹配恰好一个词order.*匹配order.pay不匹配order.pay.success#匹配零个或多个词order.#匹配order.pay、order.pay.success适合场景灵活的消息分类比如按模块和操作进行多级分类。六、五种模式对比模式交换机消费者数消息去向简单模式无直连队列1个只有一个消费者收到工作队列无直连队列多个轮询分配每人一条发布订阅Fanout多个每个队列都收到一份路由模式Direct多个只有路由键匹配的队列收到主题模式Topic多个通配符匹配的队列收到七、Day 1 小结RabbitMQ 是消息队列负责在系统之间异步传递消息三个核心作用异步处理、系统解耦、流量削峰消息流转生产者 → 交换机 → 队列 → 消费者四种交换机Direct精确、Fanout广播、Topic通配符、Headers很少用五种模式简单、工作队列、发布订阅、路由、主题八、Day 2Spring Boot 集成 RabbitMQ1. 添加依赖dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-amqp/artifactId/dependency2. 配置连接信息spring:rabbitmq:host:localhostport:5672username:guestpassword:guestvirtual-host:/virtual-host是虚拟主机类似 Redis 的 database用来隔离不同业务的消息默认用/。3. 发送消息RabbitTemplateAutowiredprivateRabbitTemplaterabbitTemplate;// 发送消息到指定队列简单模式rabbitTemplate.convertAndSend(队列名,消息内容);// 发送到交换机带路由键rabbitTemplate.convertAndSend(交换机名,路由键,消息内容);convertAndSend会自动把 Java 对象转成字节数组发送。4. 接收消息RabbitListenerRabbitListener(queues队列名)publicvoidreceiveMessage(Stringmessage){System.out.println(收到消息message);}加上RabbitListener注解后只要队列里有消息Spring 会自动调用这个方法。5. 配置队列和交换机Java 配置类ConfigurationpublicclassRabbitMQConfig{// 声明队列BeanpublicQueuetestQueue(){returnnewQueue(test_queue,true);// true持久化}// 声明 Direct 交换机BeanpublicDirectExchangedirectExchange(){returnnewDirectExchange(direct_exchange);}// 绑定队列绑定到交换机指定路由键BeanpublicBindingbinding(QueuetestQueue,DirectExchangedirectExchange){returnBindingBuilder.bind(testQueue).to(directExchange).with(test_routing_key);}}这段代码做了三件事创建一个叫test_queue的队列创建一个叫direct_exchange的 Direct 交换机把队列绑定到交换机路由键是test_routing_key6. 四种交换机类型代码对比Direct 交换机精确匹配BeanpublicDirectExchangedirectExchange(){returnnewDirectExchange(direct_exchange);}// 发送时指定路由键rabbitTemplate.convertAndSend(direct_exchange,error,错误消息);// 只有绑定了 error 的队列能收到Fanout 交换机广播BeanpublicFanoutExchangefanoutExchange(){returnnewFanoutExchange(fanout_exchange);}// 发送时不需要路由键rabbitTemplate.convertAndSend(fanout_exchange,,广播消息);// 所有绑定的队列都能收到Topic 交换机通配符BeanpublicTopicExchangetopicExchange(){returnnewTopicExchange(topic_exchange);}// 发送rabbitTemplate.convertAndSend(topic_exchange,order.pay,支付消息);// 绑定了 order.* 和 order.# 的队列都能收到7. 消息确认机制生产者确认— 消息有没有成功到达 RabbitMQspring:rabbitmq:publisher-confirm-type:correlated# 开启确认消费者确认— 消费者有没有成功处理消息spring:rabbitmq:listener:simple:acknowledge-mode:manual# 手动确认更安全手动确认代码RabbitListener(queuestest_queue)publicvoidreceiveMessage(Messagemessage,Channelchannel)throwsIOException{try{// 处理消息System.out.println(收到newString(message.getBody()));// 确认消息第二个参数false只确认当前这条channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);}catch(Exceptione){// 处理失败拒绝消息第三个参数true重回队列channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,true);}}两种确认方式对比方式配置特点自动确认acknowledge-mode: auto默认收到就确认快但可能丢消息手动确认acknowledge-mode: manual处理完再确认安全但代码复杂九、Day 2 小结Spring Boot 集成引入spring-boot-starter-amqp配置 host/port/username/password发送消息rabbitTemplate.convertAndSend(交换机, 路由键, 消息)接收消息RabbitListener(queues 队列名)自动监听配置类声明队列、交换机、绑定关系四种交换机Direct 精确、Fanout 广播、Topic 通配符、Headers 很少用消息确认生产者确认 消费者手动确认防止消息丢失十、Day 3死信队列 延迟队列 消息幂等性1. 死信队列DLX什么是死信消息死了原因有三种消息被消费者拒绝basicNack 且 requeuefalse消息过期了设置了 TTL 但没人消费队列满了新消息进不来死信队列的作用给死掉的消息一个归宿而不是直接丢掉。配置方式BeanpublicQueuenormalQueue(){MapString,ObjectargsnewHashMap();// 指定死信交换机args.put(x-dead-letter-exchange,dead_exchange);// 指定死信路由键args.put(x-dead-letter-routing-key,dead_routing_key);returnnewQueue(normal_queue,true,false,false,args);}BeanpublicQueuedeadQueue(){returnnewQueue(dead_queue,true);}消息在normal_queue里死了会自动转到dead_queue。2. 延迟队列场景用户下单后30分钟未支付自动取消订单。RabbitMQ 原生没有延迟队列用TTL 死信队列实现生产者 → 发送带TTL的消息 → 延迟队列没有消费者 ↓ 30分钟后消息过期变成死信 死信交换机 → 死信队列 → 消费者检查订单状态取消未支付订单// 发送延迟消息30分钟过期rabbitTemplate.convertAndSend(delay_exchange,order_delay,orderMessage,message-{message.getMessageProperties().setExpiration(1800000);// 30分钟单位毫秒returnmessage;});3. 消息幂等性问题消费者可能收到重复消息网络重试等原因导致业务执行两次。解决方案消息带唯一 ID消费前先用 Redis 判断是否已处理。RabbitListener(queuesdead_queue)publicvoidhandleMessage(Messagemessage,Channelchannel)throwsIOException{StringmsgIdmessage.getMessageProperties().getMessageId();// 用 Redis 判断是否已处理过SETNX只有第一次能设置成功BooleanresultredisTemplate.opsForValue().setIfAbsent(msg:msgId,1,24,TimeUnit.HOURS);if(resultnull||!result){// 已经处理过直接确认跳过channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);return;}try{// 首次处理执行业务逻辑processOrder(message);channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);}catch(Exceptione){channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,true);}}setIfAbsent相当于 Redis 的SETNX只有第一次能设置成功返回 true之后都返回 false从而保证消息只处理一次。4. 常见问题排查问题原因解决方案消息堆积消费者处理太慢增加消费者数量、提高消费速度消息丢失没有确认机制或没持久化开启确认机制 队列和消息都持久化消费者挂了自动确认模式下消息被丢失改用手动确认未确认的消息重回队列重复消费网络重试导致消息被发两次消息幂等性Redis SETNX 去重十一、三天学习总结核心知识点知识点内容消息队列作用异步处理、系统解耦、流量削峰消息流转生产者 → 交换机 → 队列 → 消费者交换机类型Direct 精确、Fanout 广播、Topic 通配符Spring Boot 集成starter-amqp RabbitTemplate RabbitListener消息确认生产者确认 消费者手动确认死信队列处理失败/过期/队列满的消息延迟队列TTL 死信队列实现订单超时取消消息幂等性Redis SETNX 防止重复消费