Spring Boot集成RabbitMQ

目录

1.RabbitMQ简介

RabbitMQ是一个由Erlang语言编写的消息中间件，它遵循AMQP协议，提供了稳定可靠的消息传输服务。RabbitMQ通过其独特的架构和丰富的功能，帮助开发者解决分布式系统中的消息传递问题，提高系统的可扩展性、可靠性和响应速度。

2.添加依赖

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency>

3.配置RabbitMQ连接

在

application.properties

或

application.yml

中配置RabbitMQ服务器的连接参数：

# 定义RabbitMQ的主机地址，这里使用的是局域网内的一个IP地址
spring.rabbitmq.host=192.168.131.130

# 指定RabbitMQ的端口号，默认情况下RabbitMQ使用5672端口
spring.rabbitmq.port=5672

# 设置RabbitMQ的用户名，这里使用的是默认的用户名guest
spring.rabbitmq.username=guest

# 设置RabbitMQ的密码，这里使用的是默认的密码guest
spring.rabbitmq.password=guest

# 配置RabbitMQ的虚拟主机，这里使用的是默认的虚拟主机"/"
spring.rabbitmq.virtual-host=/

4.DirectExchange

4.1 消费者

@ConfigurationpublicclassDirectConsumer{//注册一个队列@Bean//启动多次为什么不报错？启动的时候，它会根据这个名称Direct_Q01先去查找有没有这个队列，如果有什么都不做，如果没有创建一个新的publicQueuedirectQueue(){returnQueueBuilder.durable("Direct_Q01").maxLength(100).build();}//注册交换机@BeanpublicDirectExchangedirectExchange(){//1.启动的时候，它会根据这个名称Direct_E01先去查找有没有这个交换机，如果有什么都不做，如果没有创建一个新的returnExchangeBuilder.directExchange("Direct_E01").build();}//绑定交换机与队列关系@BeanpublicBindingdirectBinding(Queue directQueue,DirectExchange directExchange){returnBindingBuilder.bind(directQueue).to(directExchange).with("RK01");}//启动一个消费者@RabbitListener(queues ="Direct_Q01")publicvoidreceiveMessage(String msg){System.out.println("Direct_Q01收到消息："+msg);}}

4.2 生产者

//放入Ioc容器@ServicepublicclassDirectProvider{@ResourceprivateRabbitTemplate rabbitTemplate;//发送消息publicvoidsend(String message){
        rabbitTemplate.convertAndSend("Direct_E01","RK01", message);}}

4.3 测试

@SpringBootTest(classes =App.class)publicclassTestDirect{@ResourceprivateDirectProvider directProvider;@TestpublicvoiddirectSendTest(){for(int i =0; i <10; i++){
            directProvider.send("我嫩爹");}}}

4.4 一个交换机对多个队列

4.5 一个队列对多个消费者

5.FanoutExchange

5.1 消费者

@ConfigurationpublicclassFanoutConsumer{//注册一个队列@BeanpublicQueuefanoutQueue(){returnQueueBuilder.durable("Fanout_Q01").maxLength(100).build();}@BeanpublicQueuefanoutQueue2(){returnQueueBuilder.durable("Fanout_Q02").maxLength(100).build();}//注册交换机@BeanpublicFanoutExchangefanoutExchange(){returnExchangeBuilder.fanoutExchange("Fanout_E01").build();}//绑定交换机与队列关系@BeanpublicBindingfanoutBinding(Queue fanoutQueue,FanoutExchange fanoutExchange){returnBindingBuilder.bind(fanoutQueue).to(fanoutExchange);}@BeanpublicBindingfanoutBinding2(Queue fanoutQueue2,FanoutExchange fanoutExchange){returnBindingBuilder.bind(fanoutQueue2).to(fanoutExchange);}//启动一个消费者@RabbitListener(queues ="Fanout_Q01")publicvoidreceiveMessage(String msg){System.out.println("Fanout_Q01收到消息："+msg);}//启动一个消费者@RabbitListener(queues ="Fanout_Q02")publicvoidreceiveMessage2(String msg){System.out.println("Fanout_Q02收到消息："+msg);}}

5.2 生产者

@ServicepublicclassFanoutProvider{@ResourceprivateRabbitTemplate rabbitTemplate;publicvoidsend(JSONObject message){
        rabbitTemplate.convertAndSend("Fanout_E01","",message.get("msg"));}}

5.3 测试

发送请求进行测试

@RestController@RequestMapping("/fanout")publicclassFanoutController{@ResourceprivateFanoutProvider fanoutProvider;@PostMapping("/send")publicvoidsend(@RequestBodyJSONObject message){
        fanoutProvider.send(message);}}

额外涉及到的一些依赖：

<!-- 封装了一些工具类  --><dependency><groupId>cn.hutool</groupId><artifactId>hutool-all</artifactId></dependency><!--   之前web请求相关注解   --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency>

6.TopicExchange

6.1 消费者

@ConfigurationpublicclassTopicConsumer{//注册一个队列@BeanpublicQueuetopicQueue(){returnQueueBuilder.durable("Topic_Q01").maxLength(100).build();}@BeanpublicQueuetopicQueue2(){returnQueueBuilder.durable("Topic_Q02").maxLength(100).build();}//注册交换机@BeanpublicTopicExchangetopicExchange(){returnExchangeBuilder.topicExchange("Topic_E01").build();}//绑定交换机与队列关系@BeanpublicBindingtopicBinding(Queue topicQueue,TopicExchange topicExchange){returnBindingBuilder.bind(topicQueue).to(topicExchange).with("#");}@BeanpublicBindingtopicBinding2(Queue topicQueue2,TopicExchange topicExchange){returnBindingBuilder.bind(topicQueue2).to(topicExchange).with("1.8.*");}//启动一个消费者@RabbitListener(queues ="Topic_Q01")publicvoidreceiveMessage(String msg){System.out.println("Topic_Q01收到消息："+msg);}//启动一个消费者@RabbitListener(queues ="Topic_Q02")publicvoidreceiveMessage2(String msg){System.out.println("Topic_Q02收到消息："+msg);}}

6.2 生产者

@ServicepublicclassTopicProvider{@ResourceprivateRabbitTemplate rabbitTemplate;publicvoidsend(JSONObject message){
        rabbitTemplate.convertAndSend("Topic_E01",message.get("routingKey").toString(),message.get("msg"));}}