Compare commits
2 Commits
06be2485a3
...
640b80bfb7
Author | SHA1 | Date |
---|---|---|
|
640b80bfb7 | |
|
fa1e93753e |
|
@ -82,5 +82,4 @@
|
||||||
</plugin>
|
</plugin>
|
||||||
</plugins>
|
</plugins>
|
||||||
</build>
|
</build>
|
||||||
|
|
||||||
</project>
|
</project>
|
||||||
|
|
|
@ -0,0 +1,48 @@
|
||||||
|
package com.zhiLian.business.config;
|
||||||
|
|
||||||
|
import org.springframework.amqp.rabbit.connection.CorrelationData;
|
||||||
|
import org.springframework.amqp.rabbit.core.RabbitTemplate;
|
||||||
|
import org.springframework.beans.factory.annotation.Autowired;
|
||||||
|
import org.springframework.stereotype.Component;
|
||||||
|
|
||||||
|
import javax.annotation.PostConstruct;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 消息发送确认配置 消息发送到交换机的确认
|
||||||
|
*/
|
||||||
|
@Component
|
||||||
|
public class ConfirmCallbackConfig implements RabbitTemplate.ConfirmCallback {
|
||||||
|
|
||||||
|
@Autowired
|
||||||
|
private RabbitTemplate rabbitTemplate;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @PostContruct是spring框架的注解,在⽅法上加该注解会在项⽬启动的时候执⾏该⽅法,也可以理解为在spring容器初始化的时候执
|
||||||
|
* @PostConstruct bean 被初始化的时候执行的方法的注解
|
||||||
|
* @PreDestory bean 被销毁的时候执行的方法的注解
|
||||||
|
*/
|
||||||
|
@PostConstruct
|
||||||
|
public void init() {
|
||||||
|
rabbitTemplate.setConfirmCallback(this);
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 交换机不管是否收到消息的一个回调方法
|
||||||
|
*
|
||||||
|
* @param correlationData 消息相关数据
|
||||||
|
* @param ack 交换机是否收到消息
|
||||||
|
* @param cause 失败原因
|
||||||
|
*/
|
||||||
|
@Override
|
||||||
|
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
|
||||||
|
if (ack) {
|
||||||
|
// 消息投递到 broker 的状态,true表示成功
|
||||||
|
System.out.println("消息发送成功!");
|
||||||
|
} else {
|
||||||
|
// 发送异常
|
||||||
|
System.out.println("发送异常原因 = " + cause);
|
||||||
|
// TODO 可以将消息 内容 以及 失败的原因 记录到 日志表中
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
}
|
|
@ -0,0 +1,53 @@
|
||||||
|
package com.zhiLian.business.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;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* RabbitAdmin是RabbitMQ的一个Java客户端库,它提供了管理RabbitMQ资源的功能。它是通过与RabbitMQ服务器进行交互来执行管理操作的。
|
||||||
|
*/
|
||||||
|
@Configuration
|
||||||
|
public class RabbitAdminConfig {
|
||||||
|
|
||||||
|
@Value("${spring.rabbitmq.host}")
|
||||||
|
private String host;
|
||||||
|
@Value("${spring.rabbitmq.username}")
|
||||||
|
private String username;
|
||||||
|
@Value("${spring.rabbitmq.password}")
|
||||||
|
private String password;
|
||||||
|
@Value("${spring.rabbitmq.virtualhost}")
|
||||||
|
private String virtualhost;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 构建 RabbitMQ的连接工厂
|
||||||
|
* @return
|
||||||
|
*/
|
||||||
|
@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;
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 自己初始化 RabbitAdmin
|
||||||
|
* @param connectionFactory
|
||||||
|
* @return
|
||||||
|
*/
|
||||||
|
@Bean
|
||||||
|
public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
|
||||||
|
RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
|
||||||
|
rabbitAdmin.setAutoStartup(true);
|
||||||
|
return rabbitAdmin;
|
||||||
|
}
|
||||||
|
}
|
|
@ -0,0 +1,34 @@
|
||||||
|
package com.zhiLian.business.config;
|
||||||
|
|
||||||
|
import org.springframework.amqp.core.ReturnedMessage;
|
||||||
|
import org.springframework.amqp.rabbit.core.RabbitTemplate;
|
||||||
|
import org.springframework.beans.factory.annotation.Autowired;
|
||||||
|
import org.springframework.stereotype.Component;
|
||||||
|
|
||||||
|
import javax.annotation.PostConstruct;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 消息发送到队列的确认 一旦消息发送到队列失败 则会执行 returnedMessage 方法
|
||||||
|
*/
|
||||||
|
@Component
|
||||||
|
public class ReturnCallbackConfig implements RabbitTemplate.ReturnsCallback {
|
||||||
|
|
||||||
|
@Autowired
|
||||||
|
private RabbitTemplate rabbitTemplate;
|
||||||
|
|
||||||
|
@PostConstruct // @PostContruct是spring框架的注解,在⽅法上加该注解会在项⽬启动的时候执⾏该⽅法,也可以理解为在spring容器初始化的时候执
|
||||||
|
public void init() {
|
||||||
|
rabbitTemplate.setReturnsCallback(this);
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 消息发送到 队列失败 执行的 方法
|
||||||
|
* @param returnedMessage the returned message and metadata.
|
||||||
|
*/
|
||||||
|
@Override
|
||||||
|
public void returnedMessage(ReturnedMessage returnedMessage) {
|
||||||
|
System.out.println("消息" + returnedMessage.getMessage().toString() + "被交换机" + returnedMessage.getExchange() + "回退!"
|
||||||
|
+ "退回原因为:" + returnedMessage.getReplyText());
|
||||||
|
// 回退了所有的信息,可做补偿机制 记录到 数据库
|
||||||
|
}
|
||||||
|
}
|
|
@ -203,6 +203,8 @@ public class BusinessServiceImpl extends ServiceImpl<BusinessMapper, Business>
|
||||||
entinfoService.insertEntinfo(build);
|
entinfoService.insertEntinfo(build);
|
||||||
rabbitTemplate.convertAndSend("zhiLian-vehicle-exchange",JSON.toJSONString(build),message ->{
|
rabbitTemplate.convertAndSend("zhiLian-vehicle-exchange",JSON.toJSONString(build),message ->{
|
||||||
message.getMessageProperties().setMessageId(UUID.randomUUID().toString());
|
message.getMessageProperties().setMessageId(UUID.randomUUID().toString());
|
||||||
|
//设置消息延迟时间为5秒
|
||||||
|
message.getMessageProperties().setDelay(5000);
|
||||||
return message;
|
return message;
|
||||||
} );
|
} );
|
||||||
}
|
}
|
||||||
|
|
|
@ -25,7 +25,6 @@ import org.apache.commons.lang3.builder.ToStringStyle;
|
||||||
@TableName("entinfo")
|
@TableName("entinfo")
|
||||||
public class Entinfo extends BaseEntity
|
public class Entinfo extends BaseEntity
|
||||||
{
|
{
|
||||||
|
|
||||||
private static final long serialVersionUID = 1L;
|
private static final long serialVersionUID = 1L;
|
||||||
|
|
||||||
/** 数据源key */
|
/** 数据源key */
|
||||||
|
|
|
@ -39,9 +39,6 @@ import java.util.Map;
|
||||||
public class ManyDataSource {
|
public class ManyDataSource {
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
@Autowired
|
@Autowired
|
||||||
private RedisTemplate<String,String> redisTemplate;
|
private RedisTemplate<String,String> redisTemplate;
|
||||||
|
|
||||||
|
|
Loading…
Reference in New Issue