package com.newyetai.mq.consumer; import com.google.common.collect.Lists; import com.newyetai.common.constant.CommonConstants; import com.newyetai.common.util.EnvironmentUtils; import com.newyetai.mq.config.ExchangeManager; import com.newyetai.mq.config.QueueManager; import com.newyetai.mq.config.RabbitCustomProperties; import com.newyetai.mq.constant.MqConstant; import com.newyetai.mq.consumer.handler.IMqHandler; import org.springframework.beans.factory.InitializingBean; import org.springframework.beans.factory.annotation.Autowired; import java.util.List; /** * 初始化消息消费处理类,初始queue/binding * * @author sr on 2020-07-24 */ public class InitHandler implements InitializingBean { @Autowired(required = false) private List> handlers = Lists.newArrayList(); @Autowired private QueueManager queueManager; @Autowired private ExchangeManager exchangeManager; @Autowired private RabbitCustomProperties rabbitCustomProperties; @Override public void afterPropertiesSet() throws Exception { String applicationName = EnvironmentUtils.getApplicationName(); handlers.forEach(handler -> { String routingKey = handler.getRoutingKey(); String queueName = handler.getQueueName(); // 初始化binding if (queueName == null) { queueName = String.join(CommonConstants.UNDERLINE, applicationName, EnvironmentUtils.getProfile(), routingKey); if (handler.isBroadcast()) { // queueName = String.join(CommonConstants.UNDERLINE, queueName,rabbitCustomProperties.getBroadcastId()); String nacosDiscoveryIp = EnvironmentUtils.getProperty(MqConstant.NACOS_DISCOVERY_IP); queueName = String.join(CommonConstants.UNDERLINE, queueName, nacosDiscoveryIp); } } queueManager.addQueueAndRoutingKey( exchangeManager.getExchangeByType(handler.getExchangeType()), queueName, routingKey); }); } }