SpringBoot整合RabbitMQ
对于RabbitMQ开发,Spring也提供了⼀些便利.Spring和RabbitMQ的官⽅⽂档对此均有介绍.
Spring官方:Spring AMQP
RabbitMQ官方:RabbitMQ tutorial - "Hello World!" | RabbitMQ
基于SpringBoot进行RabbitMQ的开发
引入依赖
编写yml配置
#配置RabbitMQ的基本信息
spring:
rabbitmq:
host:
port: 5672 #默认为5672
username:
password:
virtual-host:
#amqp://username:password@Ip:port/virtual-host
spring:
rabbitmq:
addresses: amqp://username:password@Ip:port/virtual-host
上述两种格式均可.(自己填入具体的信息)
编写生产者代码
为了方便测试,通过接口的方式来发送信息.
为了程序的可读性和可维护性,将需要用到的一些名称通过常量类的方式声明.
声明队列,声明交换机,声明交换机和队列的绑定关系
/*** 用于声明队列,交换机和绑定关系*/
@Configuration
public class RabbitMQConfig {//工作队列模式@Bean("workQueue")public Queue workQueue(){return QueueBuilder.durable(Constants.WORK_QUEUE).build();}//发布订阅模式@Bean("fanoutQueue1")public Queue fanoutQueue1(){return QueueBuilder.durable(Constants.FANOUT_QUEUE1).build();}@Bean("fanoutQueue2")public Queue fanoutQueue2(){return QueueBuilder.durable(Constants.FANOUT_QUEUE2).build();}//声明交换机@Bean("fanoutExchange")public FanoutExchange fanoutExchange(){return ExchangeBuilder.fanoutExchange(Constants.FANOUT_EXCHANGE).durable(true).build();}//设置绑定关系@Bean("fanoutQueueBinding1")public Binding fanoutQueueBinding1(@Qualifier("fanoutQueue1") Queue queue,@Qualifier("fanoutExchange") FanoutExchange fanoutExchange){return BindingBuilder.bind(queue).to(fanoutExchange);}@Bean("fanoutQueueBinding2")public Binding fanoutQueueBinding2(@Qualifier("fanoutQueue2") Queue queue,@Qualifier("fanoutExchange") FanoutExchange fanoutExchange){return BindingBuilder.bind(queue).to(fanoutExchange);}//路由模式@Bean("directQueue1")public Queue directQueue1(){return QueueBuilder.durable(Constants.DIRECT_QUEUE1).build();}@Bean("directQueue2")public Queue directQueue2(){return QueueBuilder.durable(Constants.DIRECT_QUEUE2).build();}@Bean("directExchange")public DirectExchange directExchange(){return ExchangeBuilder.directExchange(Constants.DIRECT_EXCHANGE).durable(true).build();}//设置交换机和队列的绑定关系,带有bindingKey@Bean("directQueueBinding1")public Binding directQueueBinding1(@Qualifier("directQueue1") Queue queue,@Qualifier("directExchange") DirectExchange directExchange){return BindingBuilder.bind(queue).to(directExchange).with("orange");}@Bean("directQueueBinding2")public Binding directQueueBinding2(@Qualifier("directQueue2") Queue queue,@Qualifier("directExchange") DirectExchange directExchange){return BindingBuilder.bind(queue).to(directExchange).with("black");}@Bean("directQueueBinding3")public Binding directQueueBinding3(@Qualifier("directQueue2") Queue queue,@Qualifier("directExchange") DirectExchange directExchange){return BindingBuilder.bind(queue).to(directExchange).with("orange");}//通配符模式@Bean("topicQueue1")public Queue topicQueue1(){return QueueBuilder.durable(Constants.TOPIC_QUEUE1).build();}@Bean("topicQueue2")public Queue topicQueue2(){return QueueBuilder.durable(Constants.TOPIC_QUEUE2).build();}@Bean("topicExchange")public TopicExchange topicExchange(){return ExchangeBuilder.topicExchange(Constants.TOPIC_EXCHANGE).durable(true).build();}//绑定交换机和队列@Bean("topicQueueBinding1")public Binding topicQueueBinding1(@Qualifier("topicQueue1") Queue queue,@Qualifier("topicExchange") TopicExchange topicExchange){return BindingBuilder.bind(queue).to(topicExchange).with("*.orange.*");}@Bean("topicQueueBinding2")public Binding topicQueueBinding2(@Qualifier("topicQueue2") Queue queue,@Qualifier("topicExchange") TopicExchange topicExchange){return BindingBuilder.bind(queue).to(topicExchange).with("*.*.rabbit");}@Bean("topicQueueBinding3")public Binding topicQueueBinding3(@Qualifier("topicQueue2") Queue queue,@Qualifier("topicExchange") TopicExchange topicExchange){return BindingBuilder.bind(queue).to(topicExchange).with("lazy.#");}
}创建接口,发送消息
@RestController
@RequestMapping("/pro")
public class ProducerController {@Autowiredprivate RabbitTemplate rabbitTemplate;//工作队列@RequestMapping("/work")public String work(){for (int i = 0; i < 10; i++) {//使用内置交换机, RoutingKey 和队列名称一致rabbitTemplate.convertAndSend("", Constants.WORK_QUEUE, "hello spring amqp: work..."+i);}return "发送成功";}//发布订阅@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 routing key is "+routingKey);return "发送成功";}//通配符模式@RequestMapping("/topic/{routingKey}")public String topic(@PathVariable("routingKey") String routingKey){rabbitTemplate.convertAndSend(Constants.TOPIC_EXCHANGE, routingKey, "hello spring amqp:topic, my routing key is "+routingKey);return "发送成功";}
}编写消费者代码
关键注解
@RabbitListener
是Spring框架中⽤于监听RabbitMQ队列的注解,通过使⽤这个注解，可以定义⼀个⽅法,以便从RabbitMQ队列中接收消息.该注解⽀持多种参数类型，这些参数类型代表了从RabbitMQ接收到的消息和相关信息.
常见的参数类型:
1. String ：返回消息的内容
2. Message ( org.springframework.amqp.core.Message ):SpringAMQP的 Message 类，返回原始的消息体以及消息的属性,如消息ID,内容,队列信息等.
3. Channel ( com.rabbitmq.client.Channel )：RabbitMQ的通道对象,可以⽤于进⾏更⾼级的操作，如⼿动确认消息.
工作队列模式消费者
@Component
public class WorkListener {@RabbitListener(queues = Constants.WORK_QUEUE)public void queueListener1(Message message, Channel channel){System.out.println("listener 1 ["+Constants.WORK_QUEUE+"] 接收到消息:" +message + ",channel:"+channel);}@RabbitListener(queues = Constants.WORK_QUEUE)public void queueListener2(String message){System.out.println("listener 2 ["+Constants.WORK_QUEUE+"] 接收到消息:" +message);}
}结果
不同的参数类型,获取到的结果是不同的.
发布订阅模式消费者
@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(String message){System.out.println("队列["+Constants.FANOUT_QUEUE2+"] 接收到消息:" +message);}
}路由模式消费者
@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);}
}通配符模式消费者
@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);}
}