diff --git a/xdclass-product-service/src/main/java/net/jieyuu/config/RabbitMQConfig.java b/xdclass-product-service/src/main/java/net/jieyuu/config/RabbitMQConfig.java new file mode 100644 index 0000000..a46b8a5 --- /dev/null +++ b/xdclass-product-service/src/main/java/net/jieyuu/config/RabbitMQConfig.java @@ -0,0 +1,129 @@ +package net.jieyuu.config; + + +import lombok.Data; +import org.springframework.amqp.core.Binding; +import org.springframework.amqp.core.Exchange; +import org.springframework.amqp.core.Queue; +import org.springframework.amqp.core.TopicExchange; +import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter; +import org.springframework.amqp.support.converter.MessageConverter; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +import java.util.HashMap; +import java.util.Map; + +@Configuration +@Data +public class RabbitMQConfig { + + /** + * 交换机 + */ + @Value("${mqconfig.stock_event_exchange}") + private String eventExchange; + + + /** + * 第一个队列延迟队列, + */ + @Value("${mqconfig.stock_release_delay_queue}") + private String stockReleaseDelayQueue; + + /** + * 第一个队列的路由key + * 进入队列的路由key + */ + @Value("${mqconfig.stock_release_delay_routing_key}") + private String stockReleaseDelayRoutingKey; + + + /** + * 第二个队列,被监听恢复库存的队列 + */ + @Value("${mqconfig.stock_release_queue}") + private String stockReleaseQueue; + + /** + * 第二个队列的路由key + * 即进入死信队列的路由key + */ + @Value("${mqconfig.stock_release_routing_key}") + private String stockReleaseRoutingKey; + + /** + * 过期时间 + */ + @Value("${mqconfig.ttl}") + private Integer ttl; + + /** + * 消息转换器 + * + * @return + */ + @Bean + public MessageConverter messageConverter() { + return new Jackson2JsonMessageConverter(); + } + + /** + * 自动创建交换机 Topic类型 也可dirct交换机 + * 一个微服务一个交换机 + * + * @return + */ + @Bean + public Exchange stockEventExchange() { + return new TopicExchange(eventExchange, true, false); + } + + /** + * 延迟队列 + * + * @return + */ + @Bean + public Queue stockReleaseDelayQueue() { + Map args = new HashMap<>(3); + args.put("x-message-ttl", ttl); + args.put("x-dead-letter-exchange", eventExchange); + args.put("x-dead-letter-routing-key", stockReleaseRoutingKey); + + return new Queue(stockReleaseDelayQueue, true, false, false, args); + } + + /** + * 死信队列 普通队列,用于被监听 + * + * @return + */ + @Bean + public Queue stockReleaseQueue() { + return new Queue(stockReleaseQueue, true, false, false); + } + + /** + * 延迟队列建立绑定关系 + * + * @return + */ + @Bean + public Binding stockReleaseDelayBinding() { + return new Binding(stockReleaseDelayQueue, Binding.DestinationType.QUEUE, eventExchange, stockReleaseDelayRoutingKey, null); + } + + /** + * 死信队列建立绑定关系 + * + * @return + */ + @Bean + public Binding stockReleaseBinding() { + return new Binding(stockReleaseQueue, Binding.DestinationType.QUEUE, eventExchange, stockReleaseRoutingKey, null); + } + + +} diff --git a/xdclass-product-service/src/main/resources/application.yml b/xdclass-product-service/src/main/resources/application.yml index 445b681..37f751b 100644 --- a/xdclass-product-service/src/main/resources/application.yml +++ b/xdclass-product-service/src/main/resources/application.yml @@ -44,3 +44,22 @@ logging: root: INFO +#自定义消息队列配置,发送锁定库存消息-》延迟exchange-》lock.queue-》死信exchange-》release.queue +mqconfig: + #延迟队列,不能被监听消费 + stock_release_delay_queue: stock.release.delay.queue + + #延迟队列的消息过期后转发的队列 + stock_release_queue: stock.release.queue + + #交换机 + stock_event_exchange: stock.event.exchange + + #进入延迟队列的路由key + stock_release_delay_routing_key: stock.release.delay.routing.key + + #消息过期,进入释放队列的key + stock_release_routing_key: stock.release.routing.key + + #消息过期时间,毫秒,测试改为15秒 + ttl: 15000