RabbitMQ配置文件相关

pcloud-common-core/src/main/java/com/pcloud/common/core/constant/MQQueueConstant.java

+/**
+ * 
+ */
+package com.pcloud.common.core.constant;
+
+/**
+ * 
+ * @author：songx
+ * @date：2018年8月16日,下午9:54:51
+ */
+public class MQQueueConstant {
+
+	/**
+	 * 模板消息
+	 */
+	public static final String TEMPLATE = "templateQueue";
+
+	/**
+	 * 文件转码(PDF、OFFICE)
+	 */
+	public static final String TRANSCODE = "transcodeQueue";
+
+	/**
+	 * 文件转码(音视频)
+	 */
+	public static final String CONVERT = "convertQueue";
+
+	/**
+	 * 文件转码结束(音视频)->应用中心
+	 */
+	public static final String CONVERT_TO_APP = "convert2AppQueue";
+
+	/**
+	 * 文件转码结束(音视频)->内容中心
+	 */
+	public static final String CONVERT_TO_CONTENT = "convert2ContentQueue";
+
+	/**
+	 * 数据埋点
+	 */
+	public static final String FRONT_EVENT_QUEUE = "frontEventQueue";
+
+	/**
+	 * 死信队列名称(勿改)
+	 */
+	public static final String DEAD = "rays.dlq";
+
+}

pcloud-common-core/src/main/java/com/pcloud/common/core/constant/MQTopicProducer.java

+/**
+ * 
+ */
+package com.pcloud.common.core.constant;
+
+/**
+ * RabbitMQ 交换机名称常量类
+ * 
+ * @author：songx
+ * @date：2018年8月17日,上午10:48:18
+ */
+public class MQTopicProducer {
+
+	/**
+	 * 应用新增
+	 */
+	public static final String APP_ADD = "topic.appAdd";
+
+	/**
+	 * 文件转码结束(PDF、EXCEL)
+	 */
+	public static final String FILE_TRANSCODE = "topic.fileTranscode";
+
+	/**
+	 * topic交换机名称(勿改)
+	 */
+	public static final String EXCHAGE = "rays.topic";
+
+}

pcloud-common-core/src/main/java/com/pcloud/common/core/mq/ActiveMqProducer.java

-package com.pcloud.common.core.mq;
-
-import org.apache.activemq.command.ActiveMQQueue;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Qualifier;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-import org.springframework.jms.connection.CachingConnectionFactory;
-import org.springframework.jms.core.JmsTemplate;
-
-/**
- * @描述：
- * 
- * @作者：songx
- * @创建时间：2017年9月14日,上午11:40:25 @版本：1.0
- */
-@Configuration
-public class ActiveMqProducer {
-
-	@Autowired
-	private CachingConnectionFactory cachingConnectionFactory;
-
-	@Autowired(required = false)
-	@Qualifier("convertQueue")
-	private ActiveMQQueue convertQueue;
-
-	@Autowired(required = false)
-	@Qualifier("labelUsedQueue")
-	private ActiveMQQueue labelUsedQueue;
-
-	@Autowired(required = false)
-	@Qualifier("logQueue")
-	private ActiveMQQueue logQueue;
-
-	@Autowired(required = false)
-	@Qualifier("messageQueue")
-	private ActiveMQQueue messageQueue;
-
-	@Autowired(required = false)
-	@Qualifier("weektaskQueue")
-	private ActiveMQQueue weektaskQueue;
-
-	@Autowired(required = false)
-	@Qualifier("templateQueue")
-	private ActiveMQQueue templateQueue;
-
-	@Autowired(required = false)
-	@Qualifier("userActionQueue")
-	private ActiveMQQueue userActionQueue;
-
-	@Autowired(required = false)
-	@Qualifier("funnelFlowQueue")
-	private ActiveMQQueue funnelFlowQueue;
-
-	@Autowired(required = false)
-	@Qualifier("fundQueue")
-	private ActiveMQQueue fundQueue;
-
-	@Autowired(required = false)
-	@Qualifier("dynamicQueue")
-	private ActiveMQQueue dynamicQueue;
-
-	@Autowired(required = false)
-	@Qualifier("exapiQueue")
-	private ActiveMQQueue exapiQueue;
-
-	@Autowired(required = false)
-	@Qualifier("frontEventQueue")
-	private ActiveMQQueue frontEventQueue;
-
-	@Autowired(required = false)
-	@Qualifier("transcodeQueue")
-	private ActiveMQQueue transcodeQueue;
-
-	@Autowired(required = false)
-	@Qualifier("promotionTaskQueue")
-	private ActiveMQQueue promotionTaskQueue;
-
-	@Autowired(required = false)
-	@Qualifier("promotionTerminateQueue")
-	private ActiveMQQueue promotionTerminateQueue;
-
-	/**
-	 * 获取音视频转码JMS
-	 *
-	 * @return
-	 */
-	@Bean(name = "jmsConvertTemplate")
-	public JmsTemplate getConvertTemplate() {
-		JmsTemplate jmsTemplate = getJmsTemplate();
-		jmsTemplate.setDefaultDestination(convertQueue);
-		return jmsTemplate;
-	}
-
-	/**
-	 * 获取标签使用JMS
-	 *
-	 * @return
-	 */
-	@Bean(name = "jmsLabelUsedTemplate")
-	public JmsTemplate getLabelUsedTemplate() {
-		JmsTemplate jmsTemplate = getJmsTemplate();
-		jmsTemplate.setDefaultDestination(labelUsedQueue);
-		return jmsTemplate;
-	}
-
-	/**
-	 * 获取日志JMS
-	 *
-	 * @return
-	 */
-	@Bean(name = "jmsLogTemplate")
-	public JmsTemplate getLogTemplate() {
-		JmsTemplate jmsTemplate = getJmsTemplate();
-		jmsTemplate.setDefaultDestination(logQueue);
-		return jmsTemplate;
-	}
-
-	/**
-	 * 获取消息JMS
-	 *
-	 * @return
-	 */
-	@Bean(name = "jmsMessageTemplate")
-	public JmsTemplate getMessageTemplate() {
-		JmsTemplate jmsTemplate = getJmsTemplate();
-		jmsTemplate.setDefaultDestination(messageQueue);
-		return jmsTemplate;
-	}
-
-	/**
-	 * 获取编辑任务JMS
-	 *
-	 * @return
-	 */
-	@Bean(name = "jmsWeektaskTemplate")
-	public JmsTemplate getWeektaskTemplate() {
-		JmsTemplate jmsTemplate = getJmsTemplate();
-		jmsTemplate.setDefaultDestination(weektaskQueue);
-		return jmsTemplate;
-	}
-
-	/**
-	 * 获取模板消息JMS
-	 *
-	 * @return
-	 */
-	@Bean(name = "jmsTemplateTemplate")
-	public JmsTemplate getTemplateTemplate() {
-		JmsTemplate jmsTemplate = getJmsTemplate();
-		jmsTemplate.setDefaultDestination(templateQueue);
-		return jmsTemplate;
-	}
-
-	/**
-	 * 获取用户行为JMS
-	 *
-	 * @return
-	 */
-	@Bean(name = "jmsUserActionTemplate")
-	public JmsTemplate getUserActionTemplate() {
-		JmsTemplate jmsTemplate = getJmsTemplate();
-		jmsTemplate.setDefaultDestination(userActionQueue);
-		return jmsTemplate;
-	}
-
-	/**
-	 * 获取漏斗流程JMS
-	 *
-	 * @return
-	 */
-	@Bean(name = "jmsFunnelFlowTemplate")
-	public JmsTemplate getFunnelFlowTemplate() {
-		JmsTemplate jmsTemplate = getJmsTemplate();
-		jmsTemplate.setDefaultDestination(funnelFlowQueue);
-		return jmsTemplate;
-	}
-
-	/**
-	 * 基金修改监控JMS
-	 *
-	 * @return
-	 */
-	@Bean(name = "jmsFundTemplate")
-	public JmsTemplate getJmsFundTemplate() {
-		JmsTemplate jmsTemplate = getJmsTemplate();
-		jmsTemplate.setDefaultDestination(fundQueue);
-		return jmsTemplate;
-	}
-
-	/**
-	 * 作者动态监控JMS
-	 *
-	 * @return
-	 */
-	@Bean(name = "jmsDynamicTemplate")
-	public JmsTemplate getJmsDynamicTemplate() {
-		JmsTemplate jmsTemplate = getJmsTemplate();
-		jmsTemplate.setDefaultDestination(dynamicQueue);
-		return jmsTemplate;
-	}
-
-	/**
-	 * 收集事件
-	 *
-	 * @return
-	 */
-	@Bean(name = "jmsFrontEventTemplate")
-	public JmsTemplate getJmsFrontEventTemplate() {
-		JmsTemplate jmsTemplate = getJmsTemplate();
-		jmsTemplate.setDefaultDestination(frontEventQueue);
-		return jmsTemplate;
-	}
-
-	/**
-	 * 对外api监控JMS
-	 *
-	 * @author wangrui
-	 * @date 10:40 2018/4/13
-	 */
-	@Bean(name = "jmsExapiTemplate")
-	public JmsTemplate getJmsExapiTemplate() {
-		JmsTemplate jmsTemplate = getJmsTemplate();
-		jmsTemplate.setDefaultDestination(exapiQueue);
-		return jmsTemplate;
-	}
-
-	/**
-	 * 文件转码JMS
-	 *
-	 * @return
-	 */
-	@Bean(name = "jmsTranscodeTemplate")
-	public JmsTemplate getJmsTranscodeTemplate() {
-		JmsTemplate jmsTemplate = getJmsTemplate();
-		jmsTemplate.setDefaultDestination(transcodeQueue);
-		return jmsTemplate;
-	}
-
-	/**
-	 * 获取促销任务JMS
-	 *
-	 * @return
-	 */
-	@Bean(name = "jmsPromotionTaskTemplate")
-	public JmsTemplate getJmsPromotionTaskTemplate() {
-		JmsTemplate jmsTemplate = getJmsTemplate();
-		jmsTemplate.setDefaultDestination(promotionTaskQueue);
-		return jmsTemplate;
-	}
-
-	/**
-	 * 获取促销终止JMS
-	 *
-	 * @return
-	 */
-	@Bean(name = "jmsPromotionTerminateTemplate")
-	public JmsTemplate getJmsPromotionTerminateTemplate() {
-		JmsTemplate jmsTemplate = getJmsTemplate();
-		jmsTemplate.setDefaultDestination(promotionTerminateQueue);
-		return jmsTemplate;
-	}
-
-	/**
-	 * 获取JMS模板
-	 *
-	 * @return
-	 */
-	private JmsTemplate getJmsTemplate() {
-		JmsTemplate jmsTemplate = new JmsTemplate(cachingConnectionFactory);
-		jmsTemplate.setReceiveTimeout(10000);
-		return jmsTemplate;
-	}
-
-}
\ No newline at end of file
+package com.pcloud.common.core.mq;
+
+import org.apache.activemq.command.ActiveMQQueue;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.jms.connection.CachingConnectionFactory;
+import org.springframework.jms.core.JmsTemplate;
+
+/**
+ * @描述：
+ * 
+ * @作者：songx
+ * @创建时间：2017年9月14日,上午11:40:25 @版本：1.0
+ */
+@Configuration
+public class ActiveMqProducer {
+
+	@Autowired
+	private CachingConnectionFactory cachingConnectionFactory;
+
+	@Autowired(required = false)
+	@Qualifier("convertQueue")
+	private ActiveMQQueue convertQueue;
+
+	@Autowired(required = false)
+	@Qualifier("labelUsedQueue")
+	private ActiveMQQueue labelUsedQueue;
+
+	@Autowired(required = false)
+	@Qualifier("logQueue")
+	private ActiveMQQueue logQueue;
+
+	@Autowired(required = false)
+	@Qualifier("messageQueue")
+	private ActiveMQQueue messageQueue;
+
+	@Autowired(required = false)
+	@Qualifier("weektaskQueue")
+	private ActiveMQQueue weektaskQueue;
+
+	@Autowired(required = false)
+	@Qualifier("templateQueue")
+	private ActiveMQQueue templateQueue;
+
+	@Autowired(required = false)
+	@Qualifier("userActionQueue")
+	private ActiveMQQueue userActionQueue;
+
+	@Autowired(required = false)
+	@Qualifier("funnelFlowQueue")
+	private ActiveMQQueue funnelFlowQueue;
+
+	@Autowired(required = false)
+	@Qualifier("fundQueue")
+	private ActiveMQQueue fundQueue;
+
+	@Autowired(required = false)
+	@Qualifier("dynamicQueue")
+	private ActiveMQQueue dynamicQueue;
+
+	@Autowired(required = false)
+	@Qualifier("exapiQueue")
+	private ActiveMQQueue exapiQueue;
+
+	@Autowired(required = false)
+	@Qualifier("frontEventQueue")
+	private ActiveMQQueue frontEventQueue;
+
+	@Autowired(required = false)
+	@Qualifier("transcodeQueue")
+	private ActiveMQQueue transcodeQueue;
+
+	@Autowired(required = false)
+	@Qualifier("promotionTaskQueue")
+	private ActiveMQQueue promotionTaskQueue;
+
+	@Autowired(required = false)
+	@Qualifier("promotionTerminateQueue")
+	private ActiveMQQueue promotionTerminateQueue;
+
+	/**
+	 * 获取音视频转码JMS
+	 *
+	 * @return
+	 */
+	@Bean(name = "jmsConvertTemplate")
+	public JmsTemplate getConvertTemplate() {
+		JmsTemplate jmsTemplate = getJmsTemplate();
+		jmsTemplate.setDefaultDestination(convertQueue);
+		return jmsTemplate;
+	}
+
+	/**
+	 * 获取标签使用JMS
+	 *
+	 * @return
+	 */
+	@Bean(name = "jmsLabelUsedTemplate")
+	public JmsTemplate getLabelUsedTemplate() {
+		JmsTemplate jmsTemplate = getJmsTemplate();
+		jmsTemplate.setDefaultDestination(labelUsedQueue);
+		return jmsTemplate;
+	}
+
+	/**
+	 * 获取日志JMS
+	 *
+	 * @return
+	 */
+	@Bean(name = "jmsLogTemplate")
+	public JmsTemplate getLogTemplate() {
+		JmsTemplate jmsTemplate = getJmsTemplate();
+		jmsTemplate.setDefaultDestination(logQueue);
+		return jmsTemplate;
+	}
+
+	/**
+	 * 获取消息JMS
+	 *
+	 * @return
+	 */
+	@Bean(name = "jmsMessageTemplate")
+	public JmsTemplate getMessageTemplate() {
+		JmsTemplate jmsTemplate = getJmsTemplate();
+		jmsTemplate.setDefaultDestination(messageQueue);
+		return jmsTemplate;
+	}
+
+	/**
+	 * 获取编辑任务JMS
+	 *
+	 * @return
+	 */
+	@Bean(name = "jmsWeektaskTemplate")
+	public JmsTemplate getWeektaskTemplate() {
+		JmsTemplate jmsTemplate = getJmsTemplate();
+		jmsTemplate.setDefaultDestination(weektaskQueue);
+		return jmsTemplate;
+	}
+
+	/**
+	 * 获取模板消息JMS
+	 *
+	 * @return
+	 */
+	@Bean(name = "jmsTemplateTemplate")
+	public JmsTemplate getTemplateTemplate() {
+		JmsTemplate jmsTemplate = getJmsTemplate();
+		jmsTemplate.setDefaultDestination(templateQueue);
+		return jmsTemplate;
+	}
+
+	/**
+	 * 获取用户行为JMS
+	 *
+	 * @return
+	 */
+	@Bean(name = "jmsUserActionTemplate")
+	public JmsTemplate getUserActionTemplate() {
+		JmsTemplate jmsTemplate = getJmsTemplate();
+		jmsTemplate.setDefaultDestination(userActionQueue);
+		return jmsTemplate;
+	}
+
+	/**
+	 * 获取漏斗流程JMS
+	 *
+	 * @return
+	 */
+	@Bean(name = "jmsFunnelFlowTemplate")
+	public JmsTemplate getFunnelFlowTemplate() {
+		JmsTemplate jmsTemplate = getJmsTemplate();
+		jmsTemplate.setDefaultDestination(funnelFlowQueue);
+		return jmsTemplate;
+	}
+
+	/**
+	 * 基金修改监控JMS
+	 *
+	 * @return
+	 */
+	@Bean(name = "jmsFundTemplate")
+	public JmsTemplate getJmsFundTemplate() {
+		JmsTemplate jmsTemplate = getJmsTemplate();
+		jmsTemplate.setDefaultDestination(fundQueue);
+		return jmsTemplate;
+	}
+
+	/**
+	 * 作者动态监控JMS
+	 *
+	 * @return
+	 */
+	@Bean(name = "jmsDynamicTemplate")
+	public JmsTemplate getJmsDynamicTemplate() {
+		JmsTemplate jmsTemplate = getJmsTemplate();
+		jmsTemplate.setDefaultDestination(dynamicQueue);
+		return jmsTemplate;
+	}
+
+	/**
+	 * 收集事件
+	 *
+	 * @return
+	 */
+	@Bean(name = "jmsFrontEventTemplate")
+	public JmsTemplate getJmsFrontEventTemplate() {
+		JmsTemplate jmsTemplate = getJmsTemplate();
+		jmsTemplate.setDefaultDestination(frontEventQueue);
+		return jmsTemplate;
+	}
+
+	/**
+	 * 对外api监控JMS
+	 *
+	 * @author wangrui
+	 * @date 10:40 2018/4/13
+	 */
+	@Bean(name = "jmsExapiTemplate")
+	public JmsTemplate getJmsExapiTemplate() {
+		JmsTemplate jmsTemplate = getJmsTemplate();
+		jmsTemplate.setDefaultDestination(exapiQueue);
+		return jmsTemplate;
+	}
+
+	/**
+	 * 文件转码JMS
+	 *
+	 * @return
+	 */
+	@Bean(name = "jmsTranscodeTemplate")
+	public JmsTemplate getJmsTranscodeTemplate() {
+		JmsTemplate jmsTemplate = getJmsTemplate();
+		jmsTemplate.setDefaultDestination(transcodeQueue);
+		return jmsTemplate;
+	}
+
+	/**
+	 * 获取促销任务JMS
+	 *
+	 * @return
+	 */
+	@Bean(name = "jmsPromotionTaskTemplate")
+	public JmsTemplate getJmsPromotionTaskTemplate() {
+		JmsTemplate jmsTemplate = getJmsTemplate();
+		jmsTemplate.setDefaultDestination(promotionTaskQueue);
+		return jmsTemplate;
+	}
+
+	/**
+	 * 获取促销终止JMS
+	 *
+	 * @return
+	 */
+	@Bean(name = "jmsPromotionTerminateTemplate")
+	public JmsTemplate getJmsPromotionTerminateTemplate() {
+		JmsTemplate jmsTemplate = getJmsTemplate();
+		jmsTemplate.setDefaultDestination(promotionTerminateQueue);
+		return jmsTemplate;
+	}
+
+	/**
+	 * 获取JMS模板
+	 *
+	 * @return
+	 */
+	private JmsTemplate getJmsTemplate() {
+		JmsTemplate jmsTemplate = new JmsTemplate(cachingConnectionFactory);
+		jmsTemplate.setReceiveTimeout(10000);
+		return jmsTemplate;
+	}
+
+}

pcloud-common-core/src/main/java/com/pcloud/common/core/mq/RabbitMQFactory.java

+package com.pcloud.common.core.mq;
+
+import java.util.HashMap;
+import java.util.Map;
+
+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.amqp.core.QueueBuilder;
+import org.springframework.amqp.core.TopicExchange;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import com.pcloud.common.core.constant.MQQueueConstant;
+import com.pcloud.common.core.constant.MQTopicProducer;
+
+/**
+ * RabbitMQ
+ * 
+ * @author：songx
+ * @date：2018年8月16日,上午11:22:49
+ */
+@Configuration
+public class RabbitMQFactory {
+
+	private static final String DEAD_LETTER_EXCHANGE = "rays.dlx";
+
+	private static final String DEAD_ROUTING_KEY = "rays.dlk";
+
+	private static TopicExchange topicExchange = null;
+
+	/**
+	 * 声明业务队列同时与死信队列绑定，当业务队列的消息失败时会转发到死信队列中在进行处理，防止信息丢失
+	 * 
+	 * @param queueName
+	 *            队列名称
+	 * @return
+	 */
+	public static Queue queueBuilder(String queueName) {
+		Map<String, Object> map = new HashMap<String, Object>();
+		map.put("x-dead-letter-exchange", DEAD_LETTER_EXCHANGE);// 设置死信交换机
+		map.put("x-dead-letter-routing-key", DEAD_ROUTING_KEY);// 设置死信routingKey
+		return QueueBuilder.durable(queueName).withArguments(map).build();
+	}
+
+	@Bean
+	public Queue deadQueue() {
+		return new Queue(MQQueueConstant.DEAD);
+	}
+
+	@Bean
+	public DirectExchange deadLetterExchange() {
+		return new DirectExchange(DEAD_LETTER_EXCHANGE, true, false);
+	}
+
+	@Bean
+	public Binding deadBinding() {
+		return BindingBuilder.bind(deadQueue()).to(deadLetterExchange()).with(DEAD_ROUTING_KEY);
+	}
+
+	/**
+	 * Topic模式下生产者与消费者到交换机的绑定
+	 * 
+	 * @param queue
+	 *            消费者队列
+	 * @param producer
+	 *            生产者名称
+	 * @return
+	 */
+	public static Binding bindingExchange(Queue queue, String producer) {
+		return BindingBuilder.bind(queue).to(getTopicExchange()).with(producer);
+	}
+
+	@Bean
+	TopicExchange topicExchange() {
+		return getTopicExchange();
+	}
+
+	/**
+	 * 
+	 * @return
+	 */
+	private static TopicExchange getTopicExchange() {
+		if (topicExchange == null) {
+			topicExchange = new TopicExchange(MQTopicProducer.EXCHAGE);
+		}
+		return topicExchange;
+	}
+
+}