SpringBoot——整合RabbitMQ收发消息 - ToB企服应用市场:ToB评测及商务社交产业平台

<?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.3.12.RELEASE</version>
<relativePath/>
</parent>
<groupId>com.study</groupId>
<artifactId>springboot_rabbitmq</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>springboot_rabbitmq</name>
<description>Demo project for Spring Boot</description>
<properties>
<java.version>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-thymeleaf</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</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>
</plugin>
</plugins>
</build>
</project>

复制代码

spring.rabbitmq.host=192.168.40.128
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=123456
# 指定队列,交换机,路由键的名称
#队列用于存储消息，生产者发送消息到队列，消费者从队列接收消息进行处理。
rabbit.queue.name=springboot.queue.test
#交换机负责将消息路由到一个或多个队列中，根据指定的路由键来确定消息的路由规则。
rabbit.exchange.name=springboot.exchange.test
#在消息发送时，会指定消息的路由键，交换机根据这个路由键来决定将消息路由到哪些队列中。
rabbit.routing.key=springboot.routingkey.test

复制代码

<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
</head>
<body>
<form action="/sendMessage" method="post">
<textarea rows="4" cols="40" name="message"></textarea>
<br><input type="submit" value="发送"/>
</form>
</body>
</html>

复制代码

package com.study.springboot_rabbitmq.config;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* 配置类: 绑定队列和交换器
*/
@Configuration
public class RabbitMQConfig {
@Value("${rabbit.queue.name}")
String queueName;
@Value("${rabbit.exchange.name}")
String exchangeName;
@Value("${rabbit.routing.key}")
String routingKey;
@Bean()
public Queue initQueue(){//创建队列
return new Queue(queueName);
}
@Bean()
public DirectExchange initDirectExchange(){//创建交换器
return new DirectExchange(exchangeName);
}
@Bean
public Binding bindingDirect(){//将队列与交换器绑定
return BindingBuilder.bind(initQueue()).to(initDirectExchange()).with(routingKey);
}
}

复制代码

package com.study.springboot_rabbitmq.service;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
/**
* 消息生产者
*/
@Service
public class RabbitMQProducer {
@Autowired
RabbitTemplate rabbitTemplate;
//读取配置文件中的交换器名称和路由键名称
@Value("${rabbit.exchange.name}")
String exchangeName;
@Value("${rabbit.routing.key}")
String routingKey;
//将消息发送给RabbitMQ的交换机,交换机通过路由键决定将消息路由到哪些队列
public void send(String message){
rabbitTemplate.convertAndSend(exchangeName,routingKey,message);
}
}

复制代码

package com.study.springboot_rabbitmq.service;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;
/**
* 消息消费者
*/
@Service
public class RabbitMQConsumer {
private static final Logger log= LoggerFactory.getLogger(RabbitMQConsumer.class);
//使用@RabbitListener监听配置文件中的队列,当收到消息后,将其打印在控制台上
@RabbitListener(queues = "${rabbit.queue.name}")
public void getMessage(String message){
log.info("消费者收到消息:{}",message);
}
}

复制代码

package com.study.springboot_rabbitmq.config;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cache.annotation.Caching;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* 配置类: 绑定队列和交换器
*/
@Configuration
public class RabbitMQConfig {
private static final Logger log= LoggerFactory.getLogger(RabbitMQConfig.class);
@Autowired
private CachingConnectionFactory connectionFactory;
@Value("${rabbit.queue.name}")
String queueName;
@Value("${rabbit.exchange.name}")
String exchangeName;
@Value("${rabbit.routing.key}")
String routingKey;
@Bean()
public Queue initQueue(){//创建队列
return new Queue(queueName);
}
@Bean()
public DirectExchange initDirectExchange(){//创建交换器
return new DirectExchange(exchangeName);
}
@Bean
public Binding bindingDirect(){//将队列与交换器绑定
return BindingBuilder.bind(initQueue()).to(initDirectExchange()).with(routingKey);
}
//RabbitMQ收到消息后,把结果反馈给服务器,服务器将打印日志
@Bean
public RabbitTemplate rabbitTemplate(){
//消息发送成功后触发确认方法
connectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
//消息发送失败后触发回调方法
connectionFactory.setPublisherReturns(true);
//通过连接工厂对象创建RabbitTemplate对象
RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
//若交换器无法匹配到指定队列,则取消发送消息
rabbitTemplate.setMandatory(true);
rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {//ack:RabbitMQ返回的应答
if(ack){
log.info("消息发送成功");
}else{
log.info("消息发送失败,原因: {}",cause);
}
}
});
rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
@Override
public void returnedMessage(Message message, int i, String s, String s1, String s2) {
log.info("消息发送失败: {}",message);
}
});
return rabbitTemplate;
}
}

复制代码

package com.study.springboot_rabbitmq;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class SpringbootRabbitmqApplication {
public static void main(String[] args) {
SpringApplication.run(SpringbootRabbitmqApplication.class, args);
}
}

复制代码