1.引入依赖:
pom.xml添加依赖
<!--核心starter，内置amqp‑client--> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency>2.配置文件application.yml
spring: rabbitmq: addresses: amqp://登录RabbitMQ的用户名:登录的密码@部署RabbitMQ服务的公网IP地址:端口/虚拟机名 listener: simple: acknowledge-mode: auto #消息接收确认(自动),默认是auto,这里是为了直观观察 #以上为必写,以下为可选 # #acknowledge-mode: manual #消息接收确认(手动) # #prefetch: 1 #消费者持有最大待消费消息数 # #retry: # #enabled: true # 开启消费者失败重试 # #initial-interval: 5000ms # 初始失败等待时长为5秒 # #max-attempts: 5 # 最大重试次数 # #publisher-confirm-type: correlated #消息发送确认 # #publisher-returns: true #设置回退3.创建一个常量类(存相关名称)和一个配置类(交换机,队列,绑定关系都写里面)
Constants类:
package com.txm.rabbitmq.constant; public class Constants { //工作模式 public static final String WORK_QUEUE="work.queue"; //发布订阅模式fanout public static final String FANOUT_QUEUE1="fanout.queue1"; public static final String FANOUT_QUEUE2="fanout.queue2"; public static final String FANOUT_EXCHANGE="fanout.exchange"; //路由模式 public static final String DIRECT_QUEUE1="direct.queue1"; public static final String DIRECT_QUEUE2="direct.queue2"; public static final String DIRECT_EXCHANGE="direct.exchange"; //通配符模式 public static final String TOPIC_QUEUE1="topic.queue1"; public static final String TOPIC_QUEUE2="topic.queue2"; public static final String TOPIC_EXCHANGE="topic.exchange"; }Config类:
package com.txm.rabbitmq.config; import com.txm.rabbitmq.constant.Constants; import org.springframework.amqp.core.*; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @Configuration public class RabbitMQConfig { //工作模式 //声明队列 @Bean("workQueue") public Queue workQueue(){ return QueueBuilder.durable(Constants.WORK_QUEUE).build(); } //发布订阅模式 //1.声明队列 @Bean("fanoutQueue1") public Queue fanoutQueue1(){ return QueueBuilder.durable(Constants.FANOUT_QUEUE1).build(); } @Bean("fanoutQueue2") public Queue fanoutQueue2(){ return QueueBuilder.durable(Constants.FANOUT_QUEUE2).build(); } //2.声明交换机 @Bean("fanoutExchange") public FanoutExchange fanoutExchange(){ return ExchangeBuilder.fanoutExchange(Constants.FANOUT_EXCHANGE).durable(true).build(); } //3.声明交换机和队列的绑定关系 @Bean("fanoutQueueBinding1") public Binding fanoutQueueBinding1(@Qualifier("fanoutExchange") FanoutExchange fanoutExchange,@Qualifier("fanoutQueue1") Queue queue){ return BindingBuilder.bind(queue).to(fanoutExchange); } @Bean("fanoutQueueBinding2") public Binding fanoutQueueBinding2(@Qualifier("fanoutExchange") FanoutExchange fanoutExchange,@Qualifier("fanoutQueue2") Queue queue){ return BindingBuilder.bind(queue).to(fanoutExchange); } //路由模式 //1.声明队列 @Bean("directQueue1") public Queue DIRECT_QUEUE1(){ return QueueBuilder.durable(Constants.DIRECT_QUEUE1).build(); } @Bean("directQueue2") public Queue DIRECT_QUEUE2(){ return QueueBuilder.durable(Constants.DIRECT_QUEUE2).build(); } //2.声明交换机 @Bean("directExchange") public DirectExchange DIRECT_EXCHANGE(){ return ExchangeBuilder.directExchange(Constants.DIRECT_EXCHANGE).durable(true).build(); } //3.声明交换机和队列的绑定关系 @Bean("directQueueBinding1") public Binding directQueueBinding1(@Qualifier("directExchange") DirectExchange directExchange,@Qualifier("directQueue1") Queue queue){ return BindingBuilder.bind(queue).to(directExchange).with("orange"); } @Bean("directQueueBinding2") public Binding directQueueBinding2(@Qualifier("directExchange") DirectExchange directExchange,@Qualifier("directQueue2") Queue queue){ return BindingBuilder.bind(queue).to(directExchange).with("black"); } @Bean("directQueueBinding3") public Binding directQueueBinding3(@Qualifier("directExchange") DirectExchange directExchange,@Qualifier("directQueue2") Queue queue){ return BindingBuilder.bind(queue).to(directExchange).with("orange"); } //通配符模式 //1.声明队列 @Bean("topicQueue1") public Queue topicQueue1(){ return QueueBuilder.durable(Constants.TOPIC_QUEUE1).build(); } @Bean("topicQueue2") public Queue topicQueue2(){ return QueueBuilder.durable(Constants.TOPIC_QUEUE2).build(); } //2.声明交换机 @Bean("topicExchange") public TopicExchange TOPIC_EXCHANGE(){ return ExchangeBuilder.topicExchange(Constants.TOPIC_EXCHANGE).durable(true).build(); } //3.声明交换机和队列的绑定关系 @Bean("topicQueueBinding1") public Binding topicQueueBinding1(@Qualifier("topicExchange") TopicExchange topicExchange,@Qualifier("topicQueue1") Queue queue){ return BindingBuilder.bind(queue).to(topicExchange).with("*.orange.*"); } @Bean("topicQueueBinding2") public Binding topicQueueBinding2(@Qualifier("topicExchange") TopicExchange topicExchange,@Qualifier("topicQueue2") Queue queue){ return BindingBuilder.bind(queue).to(topicExchange).with("*.*.rabbit"); } @Bean("topicQueueBinding3") public Binding topicQueueBinding3(@Qualifier("topicExchange") TopicExchange topicExchange,@Qualifier("topicQueue2") Queue queue){ return BindingBuilder.bind(queue).to(topicExchange).with("lazy.#"); } }4.写生产者(Controller类,发送消息):
直接注入RabbitTemplate发送消息:
package com.txm.rabbitmq.controller; import com.txm.rabbitmq.constant.Constants; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.web.bind.annotation.PathVariable; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController; @RequestMapping("/producer") @RestController public class ProducerController { @Autowired private RabbitTemplate rabbitTemplate; //工作队列模式 @RequestMapping("/work") public String work(){ for (int i = 0; i < 10; i++) { //使用内置交换机,RoutingKey和队列名称一致 rabbitTemplate.convertAndSend("", Constants.WORK_QUEUE,"hello spring amqp:Lwork ..."+i); } return "发送成功"; } //发布订阅模式(fanout交换机) @RequestMapping("/fanout") public String fanout(){ rabbitTemplate.convertAndSend(Constants.FANOUT_EXCHANGE,"","hello spring amqp:fanout..."); return "发送成功"; } //路由模式 @RequestMapping("/direct/{routingKey}") public String direct(@PathVariable("routingKey") String routingKey){ rabbitTemplate.convertAndSend(Constants.DIRECT_EXCHANGE,routingKey,"hello spring amqp:direct,my routingKey is "+routingKey); return "发送成功"; } //通配符模式 @RequestMapping("/topic/{routingKey}") public String topic(@PathVariable("routingKey") String routingKey){ rabbitTemplate.convertAndSend(Constants.TOPIC_EXCHANGE,routingKey,"hello spring amqp:topic,my routingKey is "+routingKey); return "发送成功"; } }5.写消费者(listener类,监听消息)
<1>工作队列模式-消费者:
package com.txm.rabbitmq.listener; import com.txm.rabbitmq.constant.Constants; import org.springframework.amqp.core.Message; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Component; /** * 工作队列模式-消费者 */ @Component public class WorkListener { @RabbitListener(queues=Constants.WORK_QUEUE) public void queueListener1(Message message){ System.out.println("listener 1 ["+ Constants.WORK_QUEUE+"] 接收到消息:"+message); } @RabbitListener(queues=Constants.WORK_QUEUE) public void queueListener2(Message message){ System.out.println("listener 2 ["+ Constants.WORK_QUEUE+"] 接收到消息:"+message); } }<2>发布订阅模式(广播模式)-消费者:
package com.txm.rabbitmq.listener; import com.txm.rabbitmq.constant.Constants; import org.springframework.amqp.core.Message; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Component; /** * 发布订阅模式-消费者 */ @Component public class FanoutListener { @RabbitListener(queues=Constants.FANOUT_QUEUE1) public void queueListener1(String message){ System.out.println("队列["+ Constants.FANOUT_QUEUE1+"] 接收到消息:"+message); } @RabbitListener(queues=Constants.FANOUT_QUEUE2) public void queueListener2(Message message){ System.out.println(" 队列["+ Constants.FANOUT_QUEUE2+"] 接收到消息:"+message); } }<3>路由模式-消费者:
package com.txm.rabbitmq.listener; import com.txm.rabbitmq.constant.Constants; import org.springframework.amqp.core.Message; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Component; /** * 路由模式-消费者 */ @Component public class DirectListener { @RabbitListener(queues=Constants.DIRECT_QUEUE1) public void queueListener1(String message){ System.out.println("队列["+ Constants.DIRECT_QUEUE1+"] 接收到消息:"+message); } @RabbitListener(queues=Constants.DIRECT_QUEUE2) public void queueListener2(String message){ System.out.println(" 队列["+ Constants.DIRECT_QUEUE2+"] 接收到消息:"+message); } }<4>通配符模式-消费者:
package com.txm.rabbitmq.listener; import com.txm.rabbitmq.constant.Constants; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Component; /** * 通配符模式-消费者 */ @Component public class TopicListener { @RabbitListener(queues=Constants.TOPIC_QUEUE1) public void queueListener1(String message){ System.out.println("队列["+ Constants.TOPIC_QUEUE1+"] 接收到消息:"+message); } @RabbitListener(queues=Constants.TOPIC_QUEUE2) public void queueListener2(String message){ System.out.println(" 队列["+ Constants.TOPIC_QUEUE2+"] 接收到消息:"+message); } }
