feat(): 三次修改rabbitmq配置
parent
8ac2d539f4
commit
bfa5391ae1
|
|
@ -1,6 +1,5 @@
|
|||
package com.muyu.common.rabbit.config;
|
||||
package com.muyu.common.rabbit;
|
||||
|
||||
import org.springframework.amqp.rabbit.annotation.RabbitListenerConfigurer;
|
||||
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
|
||||
import org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistrar;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
|
|
@ -9,30 +8,21 @@ import org.springframework.context.annotation.Configuration;
|
|||
import org.springframework.messaging.converter.MappingJackson2MessageConverter;
|
||||
import org.springframework.messaging.handler.annotation.support.DefaultMessageHandlerMethodFactory;
|
||||
|
||||
/**
|
||||
* @author Lenovo
|
||||
* @Description rabbitMQ的监听器配置
|
||||
*/
|
||||
|
||||
@Configuration
|
||||
public class RabbitListenerConfig implements RabbitListenerConfigurer {
|
||||
public class RabbitListenerConfigurer implements org.springframework.amqp.rabbit.annotation.RabbitListenerConfigurer {
|
||||
|
||||
static {
|
||||
// 设置为信任所有类型的反序列化，确保消息能够正确反序列化
|
||||
System.setProperty("spring.amqp.deserialization.trust.all", "true");
|
||||
}
|
||||
|
||||
/**
|
||||
* RabbitMQ连接工厂，用于创建连接
|
||||
*/
|
||||
//以下配置RabbitMQ消息服务
|
||||
@Autowired
|
||||
public ConnectionFactory connectionFactory;
|
||||
|
||||
|
||||
/**
|
||||
* 创建处理器方法工厂的bean
|
||||
*
|
||||
* @return DefaultMessageHandlerMethodFactory 实例，用于处理消息的转换和方法调用
|
||||
* 处理器方法工厂
|
||||
* @return
|
||||
*/
|
||||
@Bean
|
||||
public DefaultMessageHandlerMethodFactory handlerMethodFactory() {
|
||||
|
|
@ -42,14 +32,8 @@ public class RabbitListenerConfig implements RabbitListenerConfigurer {
|
|||
return factory;
|
||||
}
|
||||
|
||||
/**
|
||||
* 配置RabbitMQ监听器的消息处理方法工厂。
|
||||
*
|
||||
* @param rabbitListenerEndpointRegistrar 实例，用于注册监听器的配置
|
||||
*/
|
||||
@Override
|
||||
public void configureRabbitListeners(RabbitListenerEndpointRegistrar rabbitListenerEndpointRegistrar) {
|
||||
// 注册自定义的消息处理方法工厂
|
||||
rabbitListenerEndpointRegistrar.setMessageHandlerMethodFactory(handlerMethodFactory());
|
||||
}
|
||||
|
||||
|
|
@ -1,44 +0,0 @@
|
|||
package com.muyu.common.rabbit.callback;
|
||||
|
||||
import jakarta.annotation.PostConstruct;
|
||||
import jakarta.annotation.Resource;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.amqp.rabbit.connection.CorrelationData;
|
||||
import org.springframework.amqp.rabbit.core.RabbitTemplate;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
/**
|
||||
* @author Lenovo
|
||||
* @Description 消息发送到broker确认回调
|
||||
*/
|
||||
|
||||
@Slf4j
|
||||
@Component
|
||||
public class ConfirmCallback implements RabbitTemplate.ConfirmCallback {
|
||||
|
||||
@Resource
|
||||
private RabbitTemplate rabbitTemplate;
|
||||
|
||||
/**
|
||||
* 初始化
|
||||
*/
|
||||
@PostConstruct
|
||||
public void init() {
|
||||
this.rabbitTemplate.setConfirmCallback(this);
|
||||
}
|
||||
|
||||
/**
|
||||
* 回调确认方法（消息发送之后 消息无论发送成功还是失败都会执行这个回调方法）
|
||||
* @param correlationData 回调的相关数据
|
||||
* @param ack 确认结果（true表示消息已经被broker接收，false表示消息未被broker接收）
|
||||
* @param cause 失败原因（当ack为false时，表示拒绝接收消息的原因；当ack为true时，该值为空）
|
||||
*/
|
||||
@Override
|
||||
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
|
||||
if (ack) {
|
||||
log.info("消息发送到broker成功！");
|
||||
} else {
|
||||
log.info("消息发送到broker失败，失败原因：{}", cause);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -1,39 +0,0 @@
|
|||
package com.muyu.common.rabbit.callback;
|
||||
|
||||
import jakarta.annotation.PostConstruct;
|
||||
import jakarta.annotation.Resource;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.amqp.core.ReturnedMessage;
|
||||
import org.springframework.amqp.rabbit.core.RabbitTemplate;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
/**
|
||||
* @author Lenovo
|
||||
* @Description 消息发送失败时回调
|
||||
*/
|
||||
|
||||
@Slf4j
|
||||
@Component
|
||||
public class ReturnsCallback implements RabbitTemplate.ReturnsCallback {
|
||||
|
||||
@Resource
|
||||
private RabbitTemplate rabbitTemplate;
|
||||
|
||||
/**
|
||||
* 初始化
|
||||
*/
|
||||
@PostConstruct
|
||||
public void init() {
|
||||
rabbitTemplate.setReturnsCallback(this);
|
||||
}
|
||||
|
||||
/**
|
||||
* 消息发送到队列失败时执行
|
||||
* @param returnedMessage 返回的消息和元数据
|
||||
*/
|
||||
@Override
|
||||
public void returnedMessage(ReturnedMessage returnedMessage) {
|
||||
log.info("消息：{}被交换机：{}回退！回退原因：{}", returnedMessage.getMessage().toString(),
|
||||
returnedMessage.getExchange(), returnedMessage.getReplyText());
|
||||
}
|
||||
}
|
||||
|
|
@ -1,72 +0,0 @@
|
|||
package com.muyu.common.rabbit.config;
|
||||
|
||||
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
|
||||
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
|
||||
import org.springframework.amqp.rabbit.core.RabbitAdmin;
|
||||
import org.springframework.beans.factory.annotation.Value;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
|
||||
/**
|
||||
* @author Lenovo
|
||||
* @Description RabbitMQ连接和管理功能配置
|
||||
*/
|
||||
|
||||
@Configuration
|
||||
public class RabbitAdminConfig {
|
||||
|
||||
/**
|
||||
* RabbitMQ服务器的主机地址
|
||||
*/
|
||||
@Value("${spring.rabbitmq.host}")
|
||||
private String host;
|
||||
|
||||
/**
|
||||
* RabbitMQ的用户名
|
||||
*/
|
||||
@Value("${spring.rabbitmq.username}")
|
||||
private String username;
|
||||
|
||||
/**
|
||||
* RabbitMQ的密码
|
||||
*/
|
||||
@Value("${spring.rabbitmq.password}")
|
||||
private String password;
|
||||
|
||||
/**
|
||||
* RabbitMQ的虚拟主机
|
||||
*/
|
||||
@Value("${spring.rabbitmq.virtualhost}")
|
||||
private String virtualhost;
|
||||
|
||||
/**
|
||||
* 创建并初始化RabbitAdmin实例
|
||||
*
|
||||
* @return RabbitAdmin 实例
|
||||
*/
|
||||
@Bean
|
||||
public RabbitAdmin rabbitAdmin() {
|
||||
RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory());
|
||||
rabbitAdmin.setAutoStartup(true);
|
||||
return rabbitAdmin;
|
||||
}
|
||||
|
||||
/**
|
||||
* 创建RabbitMQ连接工厂
|
||||
*
|
||||
* @return ConnectionFactory 实例
|
||||
*/
|
||||
@Bean
|
||||
public ConnectionFactory connectionFactory() {
|
||||
CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
|
||||
connectionFactory.setAddresses(host);
|
||||
connectionFactory.setUsername(username);
|
||||
connectionFactory.setPassword(password);
|
||||
connectionFactory.setVirtualHost(virtualhost);
|
||||
|
||||
// 配置发送确认回调时，次配置必须配置，否则即使在RabbitTemplate配置了ConfirmCallback也不会生效
|
||||
connectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
|
||||
connectionFactory.setPublisherReturns(true);
|
||||
return connectionFactory;
|
||||
}
|
||||
}
|
||||
|
|
@ -1,26 +0,0 @@
|
|||
|
||||
package com.muyu.common.rabbit.config;
|
||||
|
||||
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
|
||||
import org.springframework.amqp.support.converter.MessageConverter;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
|
||||
/**
|
||||
* @author Lenovo
|
||||
* @Description rabbitMQ消息转换器配置
|
||||
*/
|
||||
|
||||
@Configuration
|
||||
public class RabbitMQMessageConverterConfig {
|
||||
|
||||
/**
|
||||
* 消息转换配置
|
||||
*
|
||||
* @return 消息转换器
|
||||
*/
|
||||
@Bean
|
||||
public MessageConverter jsonMessageConverter() {
|
||||
return new Jackson2JsonMessageConverter();
|
||||
}
|
||||
}
|
||||
|
|
@ -1,3 +1 @@
|
|||
com.muyu.common.rabbit.config.RabbitListenerConfig
|
||||
com.muyu.common.rabbit.config.RabbitAdminConfig
|
||||
com.muyu.common.rabbit.config.RabbitMQMessageConverterConfig
|
||||
com.muyu.common.rabbit.RabbitListenerConfigurer
|
||||
|
|
|
|||
Loading…
Reference in New Issue