1. pom依赖projectxmlnshttp://maven.apache.org/POM/4.0.0xmlns:xsihttp://www.w3.org/2001/XMLSchema-instancexsi:schemaLocationhttp://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsdmodelVersion4.0.0/modelVersionparentgroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-parent/artifactIdversion2.7.18/versionrelativePath//parentgroupIdcom.example/groupIdartifactIdspringboot-rabbitmq-intro/artifactIdversion1.0-SNAPSHOT/versionpackagingjar/packagingnamespringboot-rabbitmq-intro/nameurlhttp://maven.apache.org/urlpropertiesproject.build.sourceEncodingUTF-8/project.build.sourceEncoding/propertiesdependenciesdependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-web/artifactId/dependencydependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-amqp/artifactId/dependencydependencygroupIdorg.projectlombok/groupIdartifactIdlombok/artifactIdoptionaltrue/optional/dependencydependencygroupIdjunit/groupIdartifactIdjunit/artifactIdversion3.8.1/versionscopetest/scope/dependency/dependencies/project2. yml文件spring: rabbitmq: # 连接地址 host: 127.0.0.1 port: 5672 username: guest password: guest virtual-host: / # 生产者配置消息可靠投递 publisher-confirm-type: correlated # 开启发送确认 publisher-returns: true # 开启消息退回路由失败返回 # 消费者配置消息可靠消费 listener: simple: acknowledge-mode: manual # 手动ACK生产推荐不自动丢消息 prefetch: 1 # 每次只拿1条消息均衡消费 retry: enabled: true # 开启消费失败重试 initial-interval: 1000 # 初次重试间隔1s max-attempts: 3 # 最大重试3次 multiplier: 2 # 间隔倍数3. 消息producerpackagecom.example.message;importcom.example.config.RabbitConfig;importorg.springframework.amqp.core.Message;importorg.springframework.amqp.rabbit.connection.CorrelationData;importorg.springframework.amqp.rabbit.core.RabbitTemplate;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.stereotype.Component;importjavax.annotation.PostConstruct;importjava.util.UUID;ComponentpublicclassRabbitProducer{AutowiredprivateRabbitTemplaterabbitTemplate;// 初始化回调消息发送确认、退回监听PostConstructpublicvoidinitCallback(){// 1. 生产者确认回调消息到达交换机成功/失败rabbitTemplate.setConfirmCallback((correlationData,ack,cause)-{if(ack){System.out.println(消息投递交换机成功correlationData.getId());}else{System.err.println(消息投递交换机失败原因cause);// 业务记录失败消息定时重发}});// 2. 消息退回回调交换机存在但无匹配队列消息退回rabbitTemplate.setReturnsCallback(returned-{Messagemsgreturned.getMessage();System.err.println(消息路由失败退回newString(msg.getBody()));});}// 发送 Direct 订单消息publicvoidsendOrderMsg(StringorderJson){// 唯一ID用于追踪消息CorrelationDatacorrelationDatanewCorrelationData(UUID.randomUUID().toString());rabbitTemplate.convertAndSend(RabbitConfig.DIRECT_EXCHANGE,RabbitConfig.DIRECT_ROUTING_KEY,orderJson,correlationData);}// 广播消息 FanoutpublicvoidsendFanoutMsg(Stringcontent){rabbitTemplate.convertAndSend(RabbitConfig.FANOUT_EXCHANGE,,content);}// Topic 日志消息publicvoidsendLogMsg(StringroutingKey,Stringlog){rabbitTemplate.convertAndSend(RabbitConfig.TOPIC_EXCHANGE,routingKey,log);}}4. Consumerpackagecom.example.message;importcom.example.config.RabbitConfig;importcom.rabbitmq.client.Channel;importorg.springframework.amqp.core.Message;importorg.springframework.amqp.rabbit.annotation.RabbitListener;importorg.springframework.stereotype.Component;importjava.io.IOException;ComponentpublicclassRabbitConsumer{// 监听订单直连队列RabbitListener(queuesRabbitConfig.DIRECT_QUEUE)publicvoidconsumeOrder(Messagemessage,Channelchannel)throwsIOException{longdeliveryTagmessage.getMessageProperties().getDeliveryTag();StringmsgnewString(message.getBody());try{System.out.println(收到订单消息msg);// 模拟业务处理// orderService.handle(msg);// 手动确认消息第二个参数false只确认当前这条channel.basicAck(deliveryTag,false);}catch(Exceptione){e.printStackTrace();// 消费失败拒绝消息true 重新入队false 丢弃可配合死信队列channel.basicNack(deliveryTag,false,true);}}// 短信消费者广播队列RabbitListener(queuesRabbitConfig.SMS_QUEUE)publicvoidconsumeSms(Messagemessage,Channelchannel)throwsIOException{StringcontentnewString(message.getBody());System.out.println(短信服务收到广播content);channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);}// 邮件消费者广播队列RabbitListener(queuesRabbitConfig.MAIL_QUEUE)publicvoidconsumeMail(Messagemessage,Channelchannel)throwsIOException{StringcontentnewString(message.getBody());System.out.println(邮件服务收到广播content);channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);}// 错误日志队列RabbitListener(queuesRabbitConfig.LOG_ERROR_QUEUE)publicvoidconsumeErrorLog(Messagemessage,Channelchannel)throwsIOException{StringlognewString(message.getBody());System.out.println(捕获错误日志log);channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);}}5. Config文件packagecom.example.config;importorg.springframework.amqp.core.*;importorg.springframework.context.annotation.Bean;importorg.springframework.context.annotation.Configuration;ConfigurationpublicclassRabbitConfig{// 1.Direct 直连模式订单消息示例 publicstaticfinalStringDIRECT_EXCHANGEorder.direct.exchange;publicstaticfinalStringDIRECT_QUEUEorder.direct.queue;publicstaticfinalStringDIRECT_ROUTING_KEYorder.create;// 直连交换机BeanpublicDirectExchangedirectExchange(){// durable:true 持久化交换机重启不丢失returnExchangeBuilder.directExchange(DIRECT_EXCHANGE).durable(true).build();}// 订单队列BeanpublicQueuedirectQueue(){// 持久化队列returnQueueBuilder.durable(DIRECT_QUEUE).build();}// 绑定交换机 队列 routingKeyBeanpublicBindingdirectBinding(QueuedirectQueue,DirectExchangedirectExchange){returnBindingBuilder.bind(directQueue).to(directExchange).with(DIRECT_ROUTING_KEY);}// 2.Fanout 广播模式短信、站内信多消费者 publicstaticfinalStringFANOUT_EXCHANGEmsg.fanout.exchange;publicstaticfinalStringSMS_QUEUEmsg.sms.queue;publicstaticfinalStringMAIL_QUEUEmsg.mail.queue;BeanpublicFanoutExchangefanoutExchange(){returnExchangeBuilder.fanoutExchange(FANOUT_EXCHANGE).durable(true).build();}BeanpublicQueuesmsQueue(){returnQueueBuilder.durable(SMS_QUEUE).build();}BeanpublicQueuemailQueue(){returnQueueBuilder.durable(MAIL_QUEUE).build();}BeanpublicBindingsmsBinding(QueuesmsQueue,FanoutExchangefanoutExchange){returnBindingBuilder.bind(smsQueue).to(fanoutExchange);}BeanpublicBindingmailBinding(QueuemailQueue,FanoutExchangefanoutExchange){returnBindingBuilder.bind(mailQueue).to(fanoutExchange);}// 3.Topic 主题模式日志分级 log.info log.error publicstaticfinalStringTOPIC_EXCHANGElog.topic.exchange;publicstaticfinalStringLOG_ALL_QUEUElog.all.queue;publicstaticfinalStringLOG_ERROR_QUEUElog.error.queue;BeanpublicTopicExchangetopicExchange(){returnExchangeBuilder.topicExchange(TOPIC_EXCHANGE).durable(true).build();}BeanpublicQueuelogAllQueue(){returnQueueBuilder.durable(LOG_ALL_QUEUE).build();}BeanpublicQueuelogErrorQueue(){returnQueueBuilder.durable(LOG_ERROR_QUEUE).build();}// # 匹配所有BeanpublicBindinglogAllBinding(QueuelogAllQueue,TopicExchangetopicExchange){returnBindingBuilder.bind(logAllQueue).to(topicExchange).with(log.#);}// 精准匹配 errorBeanpublicBindinglogErrorBinding(QueuelogErrorQueue,TopicExchangetopicExchange){returnBindingBuilder.bind(logErrorQueue).to(topicExchange).with(log.error);}}6 测试controllerpackagecom.example.controller;importcom.example.message.RabbitProducer;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.web.bind.annotation.GetMapping;importorg.springframework.web.bind.annotation.RequestParam;importorg.springframework.web.bind.annotation.RestController;RestControllerpublicclassMqTestController{AutowiredprivateRabbitProducerrabbitProducer;GetMapping(/send/order)publicStringsendOrder(){rabbitProducer.sendOrderMsg({\orderId\:\10001\,\amount\:99.9});return订单消息发送成功;}GetMapping(/send/fanout)publicStringsendFanout(RequestParamStringcontent){rabbitProducer.sendFanoutMsg(content);return广播消息发送成功;}GetMapping(/send/log)publicStringsendLog(RequestParamStringlevel,RequestParamStringlog){rabbitProducer.sendLogMsg(log.level,log);return日志消息发送成功;}}7 启动项目并测试packagecom.example;importorg.springframework.boot.SpringApplication;importorg.springframework.boot.autoconfigure.SpringBootApplication;/** * Hello world! * */SpringBootApplicationpublicclassApp{publicstaticvoidmain(String[]args){SpringApplication.run(App.class,args);}}2026-08-1311:51:45.629INFO1556---[nio-8080-exec-1]o.a.c.c.C.[Tomcat].[localhost].[/]:InitializingSpringDispatcherServletdispatcherServlet2026-08-1311:51:45.629INFO1556---[nio-8080-exec-1]o.s.web.servlet.DispatcherServlet:InitializingServletdispatcherServlet2026-08-1311:51:45.629INFO1556---[nio-8080-exec-1]o.s.web.servlet.DispatcherServlet:Completedinitialization in0ms 消息投递交换机成功29af7727-2d3f-4ea4-b4af-26a43950805e 收到订单消息{orderId:10001,amount:99.9}