SpringBoot整合RabbitMQ 手動應答(簡單demo)
版本說明
- JDK 1.8
- RabbitMQ 3.7.15 Erlang 22.0
- SpringBoot 2.3.3.RELEASE
- // TODO 2021年1月8日 整理CentOS安裝RabbitMQ流程
1. 在RabbitMQ的Web管理界面,創建test隊列
參數的含義
durability:是否持久化(重啟或宕機後消息依然保存)
- durable 持久
- transient 暫時
新建maven項目。
2. pom.xml
<?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 http://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.3.3.RELEASE</version> <relativePath/> <!-- lookup parent from repository --> </parent> <groupId>com.demo</groupId> <artifactId>rabbitmq-demo</artifactId> <version>1.0.0</version> <properties> <lombok.version>1.18.12</lombok.version> </properties> <dependencies> <!--web 模塊--> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> <!-- AMQP --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency> <!-- lombok--> <dependency> <groupId>org.projectlombok</groupId> <artifactId>lombok</artifactId> <optional>true</optional> <scope>provided</scope> <version>${lombok.version}</version> </dependency> </dependencies> <build> <plugins> <plugin> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-maven-plugin</artifactId> </plugin> </plugins> </build> </project>
3. application.yaml
server: port: 20002 spring: rabbitmq: # 這裡我改瞭本地的hosts,實際地址是192.168.0.121 host: vm.com port: 5672 virtual-host: / username: admin password: admin # 開啟消息確認模式 # 消息發送到交換機確認機制,是否確認回調 # publisher-confirms: true # 是否返回回調 publisher-returns: true template: #開啟mandatory: true, basic.return方法將消息返還給生產者 mandatory: true listener: simple: # 手動應答 acknowledge-mode: manual # 最少消費者數量 concurrency: 1 # 最多消費者數量 max-concurrency: 10 # 支持重試 retry: enabled: true
端口
- 5672:RabbitMQ的通信端口
- 15672:Web管理界面端口
4. RabbitmqDemo.java
@SpringBootApplication @EnableRabbit public class RabbitmqDemoApplication { public static void main(String[] args) { SpringApplication.run(RabbitmqDemoApplication.class, args); } }
5. RabbitConfig.java
@Configuration @Slf4j public class RabbitConfig { private RabbitTemplate rabbitTemplate; @Bean public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) { rabbitTemplate = new RabbitTemplate(connectionFactory); return rabbitTemplate; } }
配置RabbitMQ的消息模板。
6. 消息生產者 produce.java
@Component public class Producer { // @Qualifier("rabbitTemplate") @Autowired private RabbitTemplate rabbitTemplate; public void send() { for (int i = 0; i < 5; i++) { System.out.println("生產者發送消息,序號為: " + i); rabbitTemplate.convertAndSend("test", String.valueOf(i)); } } }
初始化消息發送模板RabbitTemplate,@Qualifier註解用於限定具體的實現類,這裡可以不指定。
7. 消息消費者 consumer.java
消費者1和消費者2均監聽test隊列。
不同的是,消費者1收到消息後返回確認應答basicAck。
而消費者2收到消息後返回拒絕應答basicRegect,消息被消費者拒絕後重新回到test隊列中,等待下次發送給消費者。
@Component @Slf4j public class Consumer { /** * 消費者1 模擬正常處理消息的情況,消息處理完畢發送確認應答 * @param message * @param channel * @throws IOException */ @RabbitListener(queues = "test") public void process1(Message message, Channel channel) throws IOException { log.info("消費者1 接收消息: " + new String(message.getBody())); log.info("消費者1 確認應答消息:" + new String(message.getBody())); channel.basicAck(message.getMessageProperties().getDeliveryTag(), true); } /** * 消費者2 模擬處理消息出錯的情況,消費者2向rabbitmq發送拒絕應答。 * 處理失敗的消息會被重新放入ready中,再次發送給消費者,直至收到確認應答 * @param message * @param channel * @throws IOException */ @RabbitListener(queues = "test") public void process2(Message message, Channel channel) throws IOException { log.info("消費者2 接收消息:" + new String(message.getBody())); log.info("消費者2 拒絕應答消息:" + new String(message.getBody())); channel.basicReject(message.getMessageProperties().getDeliveryTag(), true); } }
8. 測試RabbitMqController.java
@RestController @RequestMapping("") public class RabbitMqController { @Autowired private Producer producer; @GetMapping("/send") public String send() { producer.send(); return "發送完成"; } }
9. 測試
使用postman或瀏覽器使用Get方法請求http://localhost:20001/send,生產者會向RabbitMQ的test隊列發送5條消息:
生產者發送消息,序號為: 0
生產者發送消息,序號為: 1
生產者發送消息,序號為: 2
生產者發送消息,序號為: 3
生產者發送消息,序號為: 4
可以看出序號為2的消息3次被消費者2接收,消費者2也3次發送拒絕應答,直到第4次才被消費者1接收,並返回確認應答。
到此這篇關於SpringBoot整合RabbitMQ 手動應答 簡單demo的文章就介紹到這瞭,更多相關SpringBoot整合RabbitMQ 內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet!
推薦閱讀:
- SpringBoot+RabbitMQ實現消息可靠傳輸詳解
- RabbitMQ 3.9.7 鏡像模式集群與Springboot 2.5.5 整合
- Springboot整合Rabbitmq之Confirm和Return機制
- RocketMQTemplate 註入失敗的解決
- 詳細聊聊RabbitMQ竟無法反序列化List問題