diff --git a/learn-shop-base/learn-shop-base-notice/src/main/java/com/billow/notice/amqp/config/AmqpRegistryBeanConfig.java b/learn-shop-base/learn-shop-base-notice/src/main/java/com/billow/notice/amqp/config/AmqpRegistryBeanConfig.java index 9cc203bc..f2259c59 100644 --- a/learn-shop-base/learn-shop-base-notice/src/main/java/com/billow/notice/amqp/config/AmqpRegistryBeanConfig.java +++ b/learn-shop-base/learn-shop-base-notice/src/main/java/com/billow/notice/amqp/config/AmqpRegistryBeanConfig.java @@ -2,7 +2,6 @@ import com.billow.notice.amqp.properties.MqSetting; import com.billow.notice.amqp.properties.NoticeMqYml; -import com.billow.notice.config.SpringContextUtil; import lombok.extern.slf4j.Slf4j; import org.springframework.amqp.core.*; import org.springframework.beans.BeansException; @@ -21,6 +20,7 @@ import java.util.HashMap; import java.util.Map; import java.util.Objects; +import java.util.Set; /** * 获取 ApplicationContext @@ -46,16 +46,19 @@ public void setApplicationContext(ApplicationContext applicationContext) throws ConfigurableApplicationContext configurableApplicationContext = (ConfigurableApplicationContext) applicationContext; // 获取bean工厂并转换为DefaultListableBeanFactory defaultListableBeanFactory = (DefaultListableBeanFactory) configurableApplicationContext.getBeanFactory(); - // 保存 applicationContext - SpringContextUtil.setApplicationContext(applicationContext); } @PostConstruct - public void afterPropertiesSet() throws Exception + public void dynamicRegistryMQ() throws Exception { - /// https://www.cnblogs.com/Chary/p/14361830.html - // 注册bean + // 获取配置文件 Map mqSettingMap = this.getMqSettingList(); + // 转换生成需要构建 bean 的数据 + Map appendMqSetting = this.convertMqSetting(mqSettingMap); + if (!CollectionUtils.isEmpty(appendMqSetting)) + { + mqSettingMap.putAll(appendMqSetting); + } log.info("====================== 开始注入 MQ 设置:{} ======================", mqSettingMap.size()); for (String key : mqSettingMap.keySet()) { @@ -63,19 +66,55 @@ public void afterPropertiesSet() throws Exception // 配置数据检查 this.checkMqSetting(key, mqSetting); // 注入队列 - this.registryQueue(mqSetting); + boolean registryQueue = this.registryQueue(mqSetting); // 注入交换机 - this.registryExchange(mqSetting.getExchangeType(), mqSetting.getExchange()); + boolean registryExchange = this.registryExchange(mqSetting); // 绑定队列和交换机 + if (!registryExchange && !registryQueue) + { + continue; + } String bindingName = this.registryBinding(key, mqSetting); log.info("exchangeName:{},queueName:{},routeKey:{},bindingName:{} 绑定成功", mqSetting.getExchange(), mqSetting.getQueue(), mqSetting.getRouteKey(), bindingName); - - System.out.println(SpringContextUtil.getBean(bindingName, Binding.class)); } log.info("====================== 结束注入 MQ 设置 ======================"); } + /** + * 转换生成需要构建 bean 的数据 + * + * @param mqSettingMap + * @return Map + * @author 千面 + * @date 2021/12/17 16:06 + */ + private Map convertMqSetting(Map mqSettingMap) + { + Map appendMqSetting = new HashMap<>(); + Set keySet = mqSettingMap.keySet(); + for (String key : keySet) + { + MqSetting mqSetting = mqSettingMap.get(key); + String dlxExchange = mqSetting.getDlxExchange(); + String dlxQueue = mqSetting.getDlxQueue(); + String dlxExchangeType = mqSetting.getDlxExchangeType(); + String dlxRouteKey = mqSetting.getDlxRouteKey(); + if (StringUtils.isEmpty(dlxExchange) || StringUtils.isEmpty(dlxQueue)) + { + continue; + } + MqSetting temp = new MqSetting(); + temp.setExchange(dlxExchange); + temp.setQueue(dlxQueue); + temp.setRouteKey(dlxRouteKey); + temp.setDlxExchangeType(dlxExchangeType); + String tempKey = key + MqConstant.SUFFIX_BINDING_DLX; + appendMqSetting.put(tempKey, temp); + } + return appendMqSetting; + } + /** * 获取 mq 的设置属性 * @@ -125,19 +164,22 @@ private String registryBinding(String mqSettingName, MqSetting mqSetting) /** * 注入交换机 * - * @param exchangeType 交换机类型 - * @param exchangeName 交换机名称 + * @param mqSetting mq 配置 * @return void * @author 千面 * @date 2021/12/15 9:16 */ - private void registryExchange(String exchangeType, String exchangeName) throws ClassNotFoundException + private boolean registryExchange(MqSetting mqSetting) throws ClassNotFoundException { + // 交换机类型 + String exchangeType = mqSetting.getExchangeType(); + // 交换机名称 + String exchangeName = mqSetting.getExchange(); boolean exchangeFlag = applicationContext.containsBean(exchangeName); if (exchangeFlag) { log.warn("交换机：{} 已经存在", exchangeName); - return; + return false; } // 交换机类型 Class exchangeClass; @@ -153,9 +195,10 @@ private void registryExchange(String exchangeType, String exchangeName) throws C exchangeClass = TopicExchange.class; break; default: - throw new ClassNotFoundException("交换机：" + exchangeName + "，没有找到对应的交换机类型:" + exchangeType); + log.error("交换机：{}，没有找到对应的交换机类型:{}", exchangeName, exchangeType); + throw new ClassNotFoundException(); } - this.registerBean(exchangeName, exchangeClass, exchangeName); + return this.registerBean(exchangeName, exchangeClass, exchangeName); } /** @@ -166,15 +209,32 @@ private void registryExchange(String exchangeType, String exchangeName) throws C * @author 千面 * @date 2021/12/15 9:16 */ - private void registryQueue(MqSetting mqSetting) + private boolean registryQueue(MqSetting mqSetting) { String queueName = mqSetting.getQueue(); boolean queueFlag = applicationContext.containsBean(queueName); if (queueFlag) { log.warn("队列：{} 已经存在", queueName); + return false; + } + // 延时队列转发的死信交换机上 + Map arguments = new HashMap<>(); + String dlxExchange = mqSetting.getDlxExchange(); + if (StringUtils.hasText(dlxExchange)) + { + arguments.put(MqConstant.X_DEAD_LETTER_EXCHANGE, dlxExchange); + arguments.put(MqConstant.X_MESSAGE_TTL, mqSetting.getMessageTtl()); + String dlxRouteKey = mqSetting.getDlxRouteKey(); + if (StringUtils.hasText(dlxRouteKey)) + { + arguments.put(MqConstant.X_DEAD_LETTER_ROUTING_KEY, dlxRouteKey); + } + mqSetting.setDurable(true); } - this.registerBean(queueName, Queue.class, queueName, mqSetting.getDurable()); + // 注册 bean + return this.registerBean(queueName, Queue.class, queueName, mqSetting.getDurable(), + false, false, arguments); } /** @@ -188,23 +248,24 @@ private void registryQueue(MqSetting mqSetting) */ private void checkMqSetting(String mqSettingName, MqSetting mqSetting) { + log.info("mqSettingName:{},mqSetting{}", mqSettingName, mqSetting); String queueName = mqSetting.getQueue(); if (StringUtils.isEmpty(queueName)) { - log.warn("MQ 配置===>>>队列名称为空,自动生成"); mqSetting.setQueue(mqSettingName + MqConstant.SUFFIX_QUEUE); + log.warn("MQ 配置{}===>>>队列名称为空,自动生成：{}", mqSetting.getQueue()); } String exchangeName = mqSetting.getExchange(); if (StringUtils.isEmpty(exchangeName)) { - log.warn("MQ 配置===>>>交换机名称为空,自动生成"); mqSetting.setExchange(mqSettingName + MqConstant.SUFFIX_EXCHANGE); + log.warn("MQ 配置===>>>交换机名称为空,自动生成：{}", mqSetting.getExchange()); } String routeKey = mqSetting.getRouteKey(); if (StringUtils.isEmpty(routeKey)) { - log.error("MQ 配置===>>>路由key名称为空,自动生成"); - mqSetting.setRouteKey(mqSettingName + MqConstant.SUFFIX_ROUTE_KEY); + mqSetting.setRouteKey(""); + log.warn("MQ 配置===>>>路由key名称为空,使用默认路由"); } } @@ -218,7 +279,7 @@ private void checkMqSetting(String mqSettingName, MqSetting mqSetting) * @author 千面 * @date 2021/12/15 8:39 */ - private void registerBean(String beanName, Class beanClass, Object... args) + private boolean registerBean(String beanName, Class beanClass, Object... args) { // 如果为空，自动生成 beanName if (StringUtils.isEmpty(beanName)) @@ -235,6 +296,7 @@ private void registerBean(String beanName, Class beanClass, Object... arg BeanDefinition beanDefinition = beanDefinitionBuilder.getRawBeanDefinition(); // 注入 bean 定义 defaultListableBeanFactory.registerBeanDefinition(beanName, beanDefinition); + return true; } } diff --git a/learn-shop-base/learn-shop-base-notice/src/main/java/com/billow/notice/amqp/config/BaseMqConfig.java b/learn-shop-base/learn-shop-base-notice/src/main/java/com/billow/notice/amqp/config/BaseMqConfig.java deleted file mode 100644 index 7beed609..00000000 --- a/learn-shop-base/learn-shop-base-notice/src/main/java/com/billow/notice/amqp/config/BaseMqConfig.java +++ /dev/null @@ -1,33 +0,0 @@ -//package com.billow.notice.amqp.config; -// -//import com.billow.notice.amqp.properties.ConfigCommonProperties; -//import com.billow.notice.amqp.properties.ExchangeProperties; -//import com.billow.notice.amqp.properties.QueueProperties; -//import com.billow.notice.amqp.properties.RouteKeyProperties; -//import org.springframework.beans.factory.annotation.Autowired; -//import org.springframework.context.annotation.Configuration; -// -///** -// * 获取 mq 参数的基类 -// * -// * @author liuyongtao -// * @create 2019-09-29 15:53 -// */ -//@Configuration -//public class BaseMqConfig { -// -// @Autowired -// private ConfigCommonProperties configCommonProperties; -// -// public QueueProperties getQueue() { -// return configCommonProperties.getMq().getQueue(); -// } -// -// public RouteKeyProperties getRouteKey() { -// return configCommonProperties.getMq().getRouteKey(); -// } -// -// public ExchangeProperties getExchange() { -// return configCommonProperties.getMq().getExchange(); -// } -//} diff --git a/learn-shop-base/learn-shop-base-notice/src/main/java/com/billow/notice/amqp/config/MqConstant.java b/learn-shop-base/learn-shop-base-notice/src/main/java/com/billow/notice/amqp/config/MqConstant.java index abdb70ff..868d15c6 100644 --- a/learn-shop-base/learn-shop-base-notice/src/main/java/com/billow/notice/amqp/config/MqConstant.java +++ b/learn-shop-base/learn-shop-base-notice/src/main/java/com/billow/notice/amqp/config/MqConstant.java @@ -15,7 +15,7 @@ public class MqConstant public final static String X_DEAD_LETTER_EXCHANGE = "x-dead-letter-exchange"; /** - * DLK：配置路由器 + * DLX：配置路由器 */ public final static String X_DEAD_LETTER_ROUTING_KEY = "x-dead-letter-routing-key"; @@ -31,6 +31,7 @@ public class MqConstant * @date 2021/12/15 10:36 */ public final static String SUFFIX_QUEUE = "Queue"; + /** * 交换机默认后缀 * @@ -38,19 +39,22 @@ public class MqConstant * @date 2021/12/15 10:36 */ public final static String SUFFIX_EXCHANGE = "Exchange"; + /** - * 路由 key 默认后缀 + * 绑定默认后缀 * * @author 千面 * @date 2021/12/15 10:36 */ - public final static String SUFFIX_ROUTE_KEY = "RouteKey"; + public final static String SUFFIX_BINDING = "Binding"; + /** - * 绑定默认后缀 + * 死信绑定默认后缀 * * @author 千面 * @date 2021/12/15 10:36 */ - public final static String SUFFIX_BINDING = "Binding"; + public final static String SUFFIX_BINDING_DLX = "Binding-DLX"; + } diff --git a/learn-shop-base/learn-shop-base-notice/src/main/java/com/billow/notice/amqp/config/MqExecuteSqlConfig.java b/learn-shop-base/learn-shop-base-notice/src/main/java/com/billow/notice/amqp/config/MqExecuteSqlConfig.java deleted file mode 100644 index 1671fd24..00000000 --- a/learn-shop-base/learn-shop-base-notice/src/main/java/com/billow/notice/amqp/config/MqExecuteSqlConfig.java +++ /dev/null @@ -1,56 +0,0 @@ -//package com.billow.notice.amqp.config; -// -//import com.billow.notice.amqp.expand.MqCommon; -//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.Autowired; -//import org.springframework.context.annotation.Bean; -//import org.springframework.context.annotation.Configuration; -// -///** -// * 执行sql 配置 -// * -// * @author liuyongtao -// * @create 2019-10-31 10:50 -// */ -//@Configuration -//public class MqExecuteSqlConfig implements MqCommon -//{ -// -// @Autowired -// private BaseMqConfig baseMqConfig; -// -// @Override -// public String getQueue() { -// return baseMqConfig.getQueue().getExecuteSql(); -// } -// -// @Override -// public String getExchange() { -// return baseMqConfig.getExchange().getExecuteSql(); -// } -// -// @Override -// public String getRouteKey() { -// return baseMqConfig.getRouteKey().getExecuteSql(); -// } -// -// @Bean -// public Queue executeSqlQueue() { -// return new Queue(this.getQueue()); -// } -// -// @Bean -// public DirectExchange executeSqlExchange() { -// return new DirectExchange(this.getExchange()); -// } -// -// @Bean -// public Binding executeSqlBinding() { -// return BindingBuilder.bind(this.executeSqlQueue()) -// .to(this.executeSqlExchange()) -// .with(this.getRouteKey()); -// } -//} diff --git a/learn-shop-base/learn-shop-base-notice/src/main/java/com/billow/notice/amqp/config/MqLogCollectConfig.java b/learn-shop-base/learn-shop-base-notice/src/main/java/com/billow/notice/amqp/config/MqLogCollectConfig.java deleted file mode 100644 index f33ff6bd..00000000 --- a/learn-shop-base/learn-shop-base-notice/src/main/java/com/billow/notice/amqp/config/MqLogCollectConfig.java +++ /dev/null @@ -1,49 +0,0 @@ -//package com.billow.notice.amqp.config; -// -//import com.billow.notice.amqp.expand.MqCommon; -//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.Autowired; -//import org.springframework.context.annotation.Bean; -//import org.springframework.context.annotation.Configuration; -// -//@Configuration -//public class MqLogCollectConfig implements MqCommon -//{ -// @Autowired -// private BaseMqConfig baseMqConfig; -// -// @Override -// public String getQueue() { -// return baseMqConfig.getQueue().getLogCollect(); -// } -// -// @Override -// public String getExchange() { -// return baseMqConfig.getExchange().getLogCollect(); -// } -// -// @Override -// public String getRouteKey() { -// return baseMqConfig.getRouteKey().getLogCollect(); -// } -// -// @Bean -// public Queue logCollectQueue() { -// return new Queue(this.getQueue()); -// } -// -// @Bean -// public DirectExchange logCollectExchange() { -// return new DirectExchange(this.getExchange()); -// } -// -// @Bean -// public Binding logCollectBinding() { -// return BindingBuilder.bind(this.logCollectQueue()) -// .to(this.logCollectExchange()) -// .with(this.getRouteKey()); -// } -//} diff --git a/learn-shop-base/learn-shop-base-notice/src/main/java/com/billow/notice/amqp/config/MqRefreshEsConfig.java b/learn-shop-base/learn-shop-base-notice/src/main/java/com/billow/notice/amqp/config/MqRefreshEsConfig.java deleted file mode 100644 index 5be74319..00000000 --- a/learn-shop-base/learn-shop-base-notice/src/main/java/com/billow/notice/amqp/config/MqRefreshEsConfig.java +++ /dev/null @@ -1,56 +0,0 @@ -//package com.billow.notice.amqp.config; -// -//import com.billow.notice.amqp.expand.MqCommon; -//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.Autowired; -//import org.springframework.context.annotation.Bean; -//import org.springframework.context.annotation.Configuration; -// -///** -// * 刷新 es 缓存 配置 -// * -// * @author liuyongtao -// * @create 2019-10-31 10:50 -// */ -//@Configuration -//public class MqRefreshEsConfig implements MqCommon -//{ -// -// @Autowired -// private BaseMqConfig baseMqConfig; -// -// @Override -// public String getQueue() { -// return baseMqConfig.getQueue().getRefreshEs(); -// } -// -// @Override -// public String getExchange() { -// return baseMqConfig.getExchange().getProduct(); -// } -// -// @Override -// public String getRouteKey() { -// return baseMqConfig.getRouteKey().getRefreshEs(); -// } -// -// @Bean -// public Queue refreshEsQueue() { -// return new Queue(this.getQueue()); -// } -// -// @Bean -// public DirectExchange refreshEsExchange() { -// return new DirectExchange(this.getExchange()); -// } -// -// @Bean -// public Binding refreshEsBinding() { -// return BindingBuilder.bind(this.refreshEsQueue()) -// .to(this.refreshEsExchange()) -// .with(this.getRouteKey()); -// } -//} diff --git a/learn-shop-base/learn-shop-base-notice/src/main/java/com/billow/notice/amqp/config/MqRunJobTestConfig.java b/learn-shop-base/learn-shop-base-notice/src/main/java/com/billow/notice/amqp/config/MqRunJobTestConfig.java deleted file mode 100644 index 3529ad66..00000000 --- a/learn-shop-base/learn-shop-base-notice/src/main/java/com/billow/notice/amqp/config/MqRunJobTestConfig.java +++ /dev/null @@ -1,56 +0,0 @@ -//package com.billow.notice.amqp.config; -// -//import com.billow.notice.amqp.expand.MqCommon; -//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.Autowired; -//import org.springframework.context.annotation.Bean; -//import org.springframework.context.annotation.Configuration; -// -///** -// * 执行sql 配置 -// * -// * @author liuyongtao -// * @create 2019-10-31 10:50 -// */ -//@Configuration -//public class MqRunJobTestConfig implements MqCommon -//{ -// -// @Autowired -// private BaseMqConfig baseMqConfig; -// -// @Override -// public String getQueue() { -// return baseMqConfig.getQueue().getRunJobTest(); -// } -// -// @Override -// public String getExchange() { -// return baseMqConfig.getExchange().getRunJobTest(); -// } -// -// @Override -// public String getRouteKey() { -// return baseMqConfig.getRouteKey().getRunJobTest(); -// } -// -// @Bean -// public Queue runJobTestQueue() { -// return new Queue(this.getQueue()); -// } -// -// @Bean -// public DirectExchange runJobTestExchange() { -// return new DirectExchange(this.getExchange()); -// } -// -// @Bean -// public Binding runJobTestBinding() { -// return BindingBuilder.bind(this.runJobTestQueue()) -// .to(this.runJobTestExchange()) -// .with(this.getRouteKey()); -// } -//} diff --git a/learn-shop-base/learn-shop-base-notice/src/main/java/com/billow/notice/amqp/config/MqSecKillOrderConfig.java b/learn-shop-base/learn-shop-base-notice/src/main/java/com/billow/notice/amqp/config/MqSecKillOrderConfig.java deleted file mode 100644 index b0b77fa2..00000000 --- a/learn-shop-base/learn-shop-base-notice/src/main/java/com/billow/notice/amqp/config/MqSecKillOrderConfig.java +++ /dev/null @@ -1,83 +0,0 @@ -//package com.billow.notice.amqp.config; -// -//import org.springframework.amqp.core.*; -//import org.springframework.beans.factory.annotation.Autowired; -//import org.springframework.context.annotation.Bean; -//import org.springframework.context.annotation.Configuration; -// -///** -// * 秒杀订单 配置 -// * -// * @author liuyongtao -// * @since 2021-8-19 17:35 -// */ -//@Configuration -//public class MqSecKillOrderConfig { -// -// @Autowired -// private BaseMqConfig baseMqConfig; -// -// /** -// * 秒杀订单交换机 -// * -// * @return {@link FanoutExchange} -// * @author liuyongtao -// * @since 2021-8-23 9:05 -// */ -// @Bean -// public FanoutExchange secKillOrderExchange() { -// return ExchangeBuilder.fanoutExchange(baseMqConfig.getExchange().getSecKillOrder()).build(); -// } -// -// //********** 秒杀订单发送订单系统 **************** -// @Bean -// public Queue secKillToCoreOrderQueue() { -// return QueueBuilder.durable(baseMqConfig.getQueue().getSecKillToCoreOrder()) -// // 指定死信交换机 -// .deadLetterExchange(baseMqConfig.getExchange().getSecKillOrderDlx()) -// .build(); -// } -// -// @Bean -// public Binding coreOrderBinding() { -// return BindingBuilder.bind(this.secKillToCoreOrderQueue()) -// .to(this.secKillOrderExchange()); -// } -// -// //********** 秒杀订单发送system系统 **************** -// @Bean -// public Queue secKillToAdminSystemQueue() { -// return QueueBuilder.durable(baseMqConfig.getQueue().getSecKillToAdminSystem()).build(); -// } -// -// @Bean -// public Binding adminSystemBinding() { -// return BindingBuilder.bind(this.secKillToAdminSystemQueue()) -// .to(this.secKillOrderExchange()); -// } -// -// -// /** -// * 秒杀订单消费失败时，死信交换机 -// * -// * @return {@link FanoutExchange} -// * @author liuyongtao -// * @since 2021-8-23 9:06 -// */ -// @Bean -// public FanoutExchange secKillOrderDlxExchange() { -// return ExchangeBuilder.fanoutExchange(baseMqConfig.getExchange().getSecKillOrderDlx()).build(); -// } -// -// //********** 秒杀订单发送订单系统（死信） **************** -// @Bean -// public Queue secKillToCoreOrderDlxQueue() { -// return QueueBuilder.durable(baseMqConfig.getQueue().getSecKillToCoreOrderDlx()).build(); -// } -// -// @Bean -// public Binding coreOrderDlxBinding() { -// return BindingBuilder.bind(this.secKillToCoreOrderDlxQueue()) -// .to(this.secKillOrderDlxExchange()); -// } -//} diff --git a/learn-shop-base/learn-shop-base-notice/src/main/java/com/billow/notice/amqp/config/MqSendMailConfig.java b/learn-shop-base/learn-shop-base-notice/src/main/java/com/billow/notice/amqp/config/MqSendMailConfig.java deleted file mode 100644 index f7b063b0..00000000 --- a/learn-shop-base/learn-shop-base-notice/src/main/java/com/billow/notice/amqp/config/MqSendMailConfig.java +++ /dev/null @@ -1,56 +0,0 @@ -//package com.billow.notice.amqp.config; -// -//import com.billow.notice.amqp.expand.MqCommon; -//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.Autowired; -//import org.springframework.context.annotation.Bean; -//import org.springframework.context.annotation.Configuration; -// -///** -// * 发送邮件MQ配置 -// * -// * @author liuyongtao -// * @create 2019-09-29 15:24 -// */ -//@Configuration -//public class MqSendMailConfig implements MqCommon -//{ -// -// @Autowired -// private BaseMqConfig baseMqConfig; -// -// @Override -// public String getQueue() { -// return baseMqConfig.getQueue().getSendMail(); -// } -// -// @Override -// public String getExchange() { -// return baseMqConfig.getExchange().getSendMail(); -// } -// -// @Override -// public String getRouteKey() { -// return baseMqConfig.getRouteKey().getSendMail(); -// } -// -// @Bean -// public Queue sendMailQueue() { -// return new Queue(this.getQueue()); -// } -// -// @Bean -// public DirectExchange sendMailExchange() { -// return new DirectExchange(this.getExchange()); -// } -// -// @Bean -// public Binding sendMailBinding() { -// return BindingBuilder.bind(this.sendMailQueue()) -// .to(this.sendMailExchange()) -// .with(this.getRouteKey()); -// } -//} diff --git a/learn-shop-base/learn-shop-base-notice/src/main/java/com/billow/notice/amqp/config/MqSendMailTTLConfig.java b/learn-shop-base/learn-shop-base-notice/src/main/java/com/billow/notice/amqp/config/MqSendMailTTLConfig.java deleted file mode 100644 index 94826bb2..00000000 --- a/learn-shop-base/learn-shop-base-notice/src/main/java/com/billow/notice/amqp/config/MqSendMailTTLConfig.java +++ /dev/null @@ -1,101 +0,0 @@ -//package com.billow.notice.amqp.config; -// -//import com.billow.notice.amqp.config.MqConstant; -//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.Autowired; -//import org.springframework.context.annotation.Bean; -//import org.springframework.context.annotation.Configuration; -// -//import java.util.HashMap; -//import java.util.Map; -// -///** -// * 发送邮件MQ配置 -// * -// * @author liuyongtao -// * @create 2019-09-29 15:24 -// */ -//@Configuration -//public class MqSendMailTTLConfig { -// -// @Autowired -// private BaseMqConfig baseMqConfig; -// -// public String getExchange() { -// return baseMqConfig.getExchange().getSendMail(); -// } -// -// // 转发队列 -// public String getQueueTrt() { -// return baseMqConfig.getQueue().getSendMailTrt(); -// } -// -// // 转发路由 -// public String getRouteKeyTrt() { -// return baseMqConfig.getRouteKey().getSendMailTrt(); -// } -// -//// public String getExchangeTrt() { -//// return baseMqConfig.getExchange().getSendMailTrt(); -//// } -// -// // 延迟队列 -// public String getQueueDlx() { -// return baseMqConfig.getQueue().getSendMailDlx(); -// } -// -// // 延迟路由 -// public String getRouteKeyDlx() { -// return baseMqConfig.getRouteKey().getSendMailDlx(); -// } -//// -//// public String getExchangeDlx() { -//// return baseMqConfig.getExchange().getSendMailDlx(); -//// } -// -// // 交换机 -// @Bean -// public DirectExchange sendMailExchangeTTL() { -// return new DirectExchange(this.getExchange(), true, false); -// } -// -// // 转发队列 -// @Bean -// public Queue sendMailQueueTrt() { -// return new Queue(this.getQueueTrt(), true); -// } -// -// // 转发队列与转发交换机绑定 -// @Bean -// public Binding sendBindingTrt() { -// return BindingBuilder.bind(this.sendMailQueueTrt()) -// .to(this.sendMailExchangeTTL()) -// .with(this.getRouteKeyTrt()); -// } -// -//// // 交换机 -//// @Bean -//// public DirectExchange sendMailExchangeDlx() { -//// return new DirectExchange(this.getExchangeDlx(), true, false); -//// } -// -// // ttl 队列 -// @Bean -// public Queue sendMailQueueDlx() { -// Map args = new HashMap<>(); -// args.put(MqConstant.X_DEAD_LETTER_EXCHANGE, this.getExchange()); -// args.put(MqConstant.X_DEAD_LETTER_ROUTING_KEY, this.getRouteKeyTrt()); -// return new Queue(this.getQueueDlx(), true, false, false, args); -// } -// -// // ttl 队列与 死信交换机 绑定 -// @Bean -// public Binding sendBindingDlx() { -// return BindingBuilder.bind(this.sendMailQueueDlx()) -// .to(this.sendMailExchangeTTL()) -// .with(this.getRouteKeyDlx()); -// } -//} diff --git a/learn-shop-base/learn-shop-base-notice/src/main/java/com/billow/notice/amqp/properties/MqSetting.java b/learn-shop-base/learn-shop-base-notice/src/main/java/com/billow/notice/amqp/properties/MqSetting.java index ac26a8ea..c469bdf3 100644 --- a/learn-shop-base/learn-shop-base-notice/src/main/java/com/billow/notice/amqp/properties/MqSetting.java +++ b/learn-shop-base/learn-shop-base-notice/src/main/java/com/billow/notice/amqp/properties/MqSetting.java @@ -10,7 +10,8 @@ * @since 2021/12/14 22:40 */ @Data -public class MqSetting { +public class MqSetting +{ /** * 交换机名称 @@ -52,4 +53,75 @@ public class MqSetting { * @since 2021/12/15 22:26 */ private Boolean durable = false; + + /** + * 延迟交换机名称 + * + * @author xiaoy + * @since 2021/12/14 22:40 + */ + private String ttlExchange; + + + /** + * 延迟队列 + * + * @author xiaoy + * @since 2021/12/14 22:40 + */ + private String ttlQueue; + + /** + * 延迟路由 + * + * @author xiaoy + * @since 2021/12/14 22:40 + */ + private String ttlRouteKey; + + /** + * 定义消息的过期时间（毫秒） + *

+ * 默认 10s + * + * @author 千面 + * @date 2021/12/17 15:54 + */ + private int messageTtl = 10 * 1000; + + /** + * 死信交换机名称（如果配置，延迟消息会转发到对应的交换机、路由、队列） + * + * @author xiaoy + * @since 2021/12/14 22:40 + */ + private String dlxExchange; + + + /** + * 死信交换机类型，默认 direct + * {@link ExchangeTypes} + * + * @author xiaoy + * @since 2021/12/14 22:47 + */ + private String dlxExchangeType = ExchangeTypes.DIRECT; + + /** + * 死信路由（如果配置，延迟消息会转发到对应的交换机、路由、队列） + * + * @author xiaoy + * @since 2021/12/14 22:40 + */ + private String dlxRouteKey; + + + /** + * 死信队列（如果配置，延迟消息会转发到对应的交换机、路由、队列） + * + * @author xiaoy + * @since 2021/12/14 22:40 + */ + private String dlxQueue; + } diff --git a/learn-shop-base/learn-shop-base-notice/src/main/java/com/billow/notice/config/SpringContextUtil.java b/learn-shop-base/learn-shop-base-notice/src/main/java/com/billow/notice/config/SpringContextUtil.java deleted file mode 100644 index 5b592529..00000000 --- a/learn-shop-base/learn-shop-base-notice/src/main/java/com/billow/notice/config/SpringContextUtil.java +++ /dev/null @@ -1,112 +0,0 @@ -package com.billow.notice.config; - -import org.springframework.beans.BeansException; -import org.springframework.beans.factory.NoSuchBeanDefinitionException; -import org.springframework.context.ApplicationContext; -import org.springframework.core.io.Resource; - -/** - * 以静态变量保存Spring ApplicationContext, 可在任何代码任何地方任何时候中取出ApplicaitonContext. - * - * @author liuyongtao - * @date 2017年4月18日 下午3:52:31 - */ -public class SpringContextUtil -{ - - private static ApplicationContext applicationContext; - - /** - * 实现ApplicationContextAware接口的context注入函数, 将其存入静态变量 - */ - public static void setApplicationContext(ApplicationContext applicationContext) throws BeansException { - SpringContextUtil.applicationContext = applicationContext; - } - - /** - * 实现ApplicationContextAware接口的context注入函数, 将其存入静态变量 - */ - public static ApplicationContext getApplicationContext() throws BeansException { - return applicationContext; - } - - /** - * 检查applicationContext是否注入 - *

- *
- * added by liuyongtao
- * - * @return - * @date 2017年4月18日 下午3:53:30 - */ - private static void checkApplicationContext() { - if (applicationContext == null) { - throw new IllegalStateException("applicaitonContext未注入,请在applicationContext.xml中定义ContextUtils"); - } - } - - /** - * 通过name和clazz返回指定的Bean - */ - public static T getBean(String name, Class clazz) { - checkApplicationContext(); - T bean = applicationContext.getBean(name, clazz); - if (bean == null) { - throw new NoSuchBeanDefinitionException(name); - } - return bean; - } - - /** - * 从静态变量ApplicationContext中取得Bean, 自动转型为所赋值对象的类型. - *

- *
- * added by liuyongtao
- * - * @param name - * @return - * @date 2017年4月18日 下午4:03:54 - */ - @SuppressWarnings("unchecked") - public static T getBean(String name) throws Exception { - checkApplicationContext(); - Object bean = applicationContext.getBean(name); - if (bean == null) { - throw new NoSuchBeanDefinitionException(name); - } - return (T) bean; - } - - /** - * 从静态变量ApplicationContext中取得Bean, 自动转型为所赋值对象的类型. - *

- *
- * added by liuyongtao
- * - * @param clazz - * @return - * @date 2017年4月18日 下午4:04:09 - */ - @SuppressWarnings("unchecked") - public static T getBean(Class clazz){ - checkApplicationContext(); - T bean = applicationContext.getBean(clazz); - if (bean == null) { - throw new NoSuchBeanDefinitionException(clazz); - } - return bean; - } - - /** - * 获取指定资源 - * - * @param name - * @return org.springframework.core.io.Resource - * @author billow - * @date 2019/8/11 12:05 - */ - public static Resource getResource(String name) { - checkApplicationContext(); - return applicationContext.getResource(name); - } -} diff --git a/learn-shop-base/learn-shop-base-notice/src/main/resources/application.yml b/learn-shop-base/learn-shop-base-notice/src/main/resources/application.yml index 66557139..03f53b8d 100644 --- a/learn-shop-base/learn-shop-base-notice/src/main/resources/application.yml +++ b/learn-shop-base/learn-shop-base-notice/src/main/resources/application.yml @@ -1,11 +1,3 @@ -##集成钉钉服务 -#ding: -# robot: -# # webhook -# webhook: https://oapi.dingtalk.com/robot/send?access_token=d1e4bf1cb3b57616d92c86ef1e9783064cde5e8e6c28944801e719fa3588bf95 -# # 加签 -# robot-key: SECb2310283956906fe7acf381ff4fce58efc12f607579160361d274f15bbf8d29f - notice: # 钉钉配置 ding: @@ -20,7 +12,7 @@ notice: # 中间件类型 type: rabbit # mq服务器地址 - host: 192.168.137.200 + host: 127.0.0.1 # mq服务器端口 port: 5672 # mq用户名和密码 @@ -36,5 +28,47 @@ notice: exchange: logCollectExchange exchangeType: direct route-key: logCollectRouteKey + # 运行自动任务 + runJobTest: + queue: runJobTestQueue + exchange: runJobTestExchange + exchangeType: direct + route-key: runJobTestRouteKey + # 执行sql 配置 + executeSql: + queue: executeSqlQueue + exchange: executeSqlExchange + route-key: executeSqlRouteKey + # 刷新 es 缓存 配置 + refreshEs: + queue: refreshEsQueue + exchange: productExchange + route-key: refreshEsRouteKey + # 秒杀订单发送订单系统 配置 + secKillToCoreOrder: + exchange: secKillOrderExchange + exchangeType: fanout + queue: secKillToCoreOrderQueue + dlx-queue: secKillToCoreOrderQueueDlx + dlx-exchange: secKillOrderExchangeDlx + dlx-exchangeType: fanout + # 秒杀订单发送system系统 配置 + secKillToAdminSystem: + exchange: secKillOrderExchange + exchangeType: fanout + queue: secKillToAdminSystemQueue + # 发送邮件 配置 + sendMail: + queue: sendmailQueue + exchange: sendMailExchange + route-key: sendmailRouteKey + + + + + + + + diff --git a/learn-shop-core-promotion/pom.xml b/learn-shop-core-promotion/pom.xml index 0c2cb9d1..50f354a1 100644 --- a/learn-shop-core-promotion/pom.xml +++ b/learn-shop-core-promotion/pom.xml @@ -35,6 +35,10 @@ com.billow learn-shop-base-aop + + com.billow + learn-shop-base-notice + com.alibaba.cloud diff --git a/learn-shop-core-promotion/src/main/java/com/billow/promotion/service/impl/SeckillServiceImpl.java b/learn-shop-core-promotion/src/main/java/com/billow/promotion/service/impl/SeckillServiceImpl.java index 634d5537..df17b4a6 100644 --- a/learn-shop-core-promotion/src/main/java/com/billow/promotion/service/impl/SeckillServiceImpl.java +++ b/learn-shop-core-promotion/src/main/java/com/billow/promotion/service/impl/SeckillServiceImpl.java @@ -1,7 +1,7 @@ package com.billow.promotion.service.impl; -import com.billow.common.amqp.config.MqSecKillOrderConfig; -import com.billow.common.amqp.expand.SendMessage; +import com.billow.notice.amqp.properties.NoticeMqYml; +import com.billow.notice.amqp.service.SendMQService; import com.billow.promotion.cache.SeckillProductCache; import com.billow.promotion.common.enums.OrderTypeEnum; import com.billow.promotion.common.enums.SeckillStatEnum; @@ -57,7 +57,9 @@ public class SeckillServiceImpl implements SeckillService { @Autowired private SeckillProductCache seckillProductCache; @Autowired - private MqSecKillOrderConfig mqSecKillOrderConfig; + private SendMQService sendMQService; + @Autowired + private NoticeMqYml noticeMqYml; @Override public ExposerVo genSeckillUrl(Long seckillProductId) { @@ -154,7 +156,8 @@ public SeckillExecutionVo executionSeckill(String md5, Long seckillProductId, St killedVo.setUsercode(userCode); killedVo.setCount(1); FieldUtils.setCommonFieldByInsert(killedVo, userCode); - SendMessage.send(mqSecKillOrderConfig.secKillOrderExchange().getName(), killedVo); + String exchange = noticeMqYml.getMqCollect().get("secKillToCoreOrder").getExchange(); + sendMQService.send(exchange, killedVo); } return new SeckillExecutionVo(seckillProductId, statEnum); }