使用SpringBoot方式开发RabbitMQ。
1、使用IDEA建立SpringBoot项目,并添加相关引用:
POM文件如下:
<?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>2.5.6</version> <relativePath/> <!-- lookup parent from repository --> </parent> <groupId>com.yas</groupId> <artifactId>RabbitMQBoot</artifactId> <version>0.0.1-SNAPSHOT</version> <name>RabbitMQBoot</name> <description>Demo project for Spring Boot</description> <properties> <java.version>1.8</java.version> </properties> <dependencies> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-devtools</artifactId> <scope>runtime</scope> <optional>true</optional> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-configuration-processor</artifactId> <optional>true</optional> </dependency> <dependency> <groupId>org.projectlombok</groupId> <artifactId>lombok</artifactId> <optional>true</optional> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-test</artifactId> <scope>test</scope> </dependency> <dependency> <groupId>org.springframework.amqp</groupId> <artifactId>spring-rabbit-test</artifactId> <scope>test</scope> </dependency> </dependencies> <build> <plugins> <plugin> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-maven-plugin</artifactId> <configuration> <excludes> <exclude> <groupId>org.projectlombok</groupId> <artifactId>lombok</artifactId> </exclude> </excludes> </configuration> </plugin> </plugins> </build> </project>
2、编写配置文件:
application.yml文件:
spring: rabbitmq: host: localhost port: 5672 username: username password: password virtual-host: /
3、编写配置类:
1 package com.yas.rabbitmqboot.config; 2 3 import org.springframework.amqp.core.Binding; 4 import org.springframework.amqp.core.BindingBuilder; 5 import org.springframework.amqp.core.Queue; 6 import org.springframework.amqp.core.TopicExchange; 7 import org.springframework.context.annotation.Bean; 8 import org.springframework.context.annotation.Configuration; 9 10 @Configuration 11 public class RabbitMQConfig { 12 //1 创建exchange - topic 13 @Bean 14 public TopicExchange getTopicExchange() { 15 return new TopicExchange("boot-topic-exchange", true, false); 16 } 17 18 //2 创建queue 19 @Bean 20 public Queue getQueue() { 21 return new Queue("boot-queue", true, false, false, null); 22 } 23 24 //3 绑定在一起 25 @Bean 26 public Binding getBinding(TopicExchange topicExchange, Queue queue) { 27 return BindingBuilder.bind(queue).to(topicExchange).with("*.red.*"); 28 } 29 }
4、生产者代码:
1 package com.yas.rabbitmqboot; 2 3 import org.junit.jupiter.api.Test; 4 import org.springframework.amqp.rabbit.core.RabbitTemplate; 5 import org.springframework.beans.factory.annotation.Autowired; 6 import org.springframework.boot.test.context.SpringBootTest; 7 8 @SpringBootTest 9 class RabbitMqBootApplicationTests { 10 @Autowired 11 RabbitTemplate rabbitTemplate; 12 13 @Test 14 void contextLoads() { 15 rabbitTemplate.convertAndSend("boot-topic-exchange","slow.red.dog","慢红狗"); 16 System.out.println("消息已生产"); 17 } 18 }
5、消费者代码:
1 package com.yas.rabbitmqboot.listen; 2 3 import org.springframework.amqp.rabbit.annotation.RabbitListener; 4 import org.springframework.stereotype.Component; 5 6 @Component 7 public class Consumer { 8 @RabbitListener(queues = "boot-queue") 9 public void getMessage(Object message) { 10 System.out.println("接收到消息:" + message); 11 } 12 }
6、手动ack的方法:
修改第2步的配置文件:
spring: rabbitmq: host: 106.12.17.17 port: 5672 username: admin password: cs1986@0312 virtual-host: / listener: simple: acknowledge-mode: manual #手动提交ack
修改第5步消费者代码:
1 package com.yas.rabbitmqboot.listen; 2 3 import com.rabbitmq.client.Channel; 4 import org.springframework.amqp.core.Message; 5 import org.springframework.amqp.rabbit.annotation.RabbitListener; 6 import org.springframework.stereotype.Component; 7 8 import java.io.IOException; 9 10 @Component 11 public class Consumer { 12 //自动ack 13 // @RabbitListener(queues = "boot-queue") 14 // public void getMessage(Object message) { 15 // System.out.println("接收到消息:" + message); 16 // } 17 18 //手动ack 19 @RabbitListener(queues = "boot-queue") 20 public void getMessage(String msg, Channel channel, Message message) throws IOException { 21 System.out.println("接收到消息:" + msg); 22 //手动ack 23 channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); 24 } 25 }