增加mq发送

Jiang
1 parent 033bee1e
Showing 11 changed files with 192 additions and 3 deletions
dtms-client/src/main/java/com/b2c/dtms/client/domain/dto/request/DtmsProduceRequestDto.java
dtms-dao/src/main/resources/sqlmap-config.xml
dtms-domain/pom.xml
dtms-rpc/pom.xml
dtms-rpc/src/main/java/com/b2c/dtms/service/rpc/MqService.java
dtms-rpc/src/main/java/com/b2c/dtms/service/rpc/impl/MqServiceImpl.java
dtms-service/src/main/java/com/b2c/dtms/handler/MqHandlerImpl.java
dtms-web/src/main/java/com/b2c/dtms/web/restful/DtmsRestController.java
dtms-web/src/main/resources/spring-rpc.xml
dtms-web/src/main/resources/spring.xml
pom.xml
@@ -26,6 +26,11 @@ public class DtmsProduceRequestDto implements Serializable {
 	private Date runtime;
 	/**
+	 * 消息主题
+	 */
+	private String topic;
+
+	/**
 	 * 消息内容
 	 */
 	private String content;
@@ -77,6 +82,14 @@ public class DtmsProduceRequestDto implements Serializable {
 		this.runtime = runtime;
 	}
+	public String getTopic() {
+		return topic;
+	}
+
+	public void setTopic(String topic) {
+		this.topic = topic;
+	}
+
 	public String getContent() {
 		return content;
 	}
@@ -124,4 +137,5 @@ public class DtmsProduceRequestDto implements Serializable {
 	public void setType(Integer type) {
 		this.type = type;
 	}
+
 }
@@ -11,7 +11,7 @@
 		<setting name="useColumnLabel" value="true" />
 		<!-- 数据库超过25000秒仍未响应则超时 -->
 		<setting name="defaultStatementTimeout" value="25000" />
-		<setting name="logImpl" value="STDOUT_LOGGING" />
+		<!-- <setting name="logImpl" value="STDOUT_LOGGING" /> -->
 	</settings>
 	<!-- 全局别名设置，在映射文件中只需写别名，而不必写出整个类路径 -->
@@ -10,6 +10,11 @@
 	<dependencies>
 		<dependency>
+			<groupId>${project.groupId}</groupId>
+			<artifactId>dtms-rpc</artifactId>
+			<version>${project.parent.version}</version>
+		</dependency>
+		<dependency>
 			<groupId>org.springframework</groupId>
 			<artifactId>spring-webmvc</artifactId>
 			<exclusions>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<modelVersion>4.0.0</modelVersion>
+	<parent>
+		<groupId>com.b2c.dtms</groupId>
+		<artifactId>dtms-parent</artifactId>
+		<version>0.0.1-SNAPSHOT</version>
+	</parent>
+	<artifactId>dtms-rpc</artifactId>
+
+	<dependencies>
+
+		<!-- 消息中心/待办事项 -->
+		<dependency>
+			<groupId>com.yqyw.messageCenter.sdk</groupId>
+			<artifactId>diligrp-messageCenter-sdk</artifactId>
+		</dependency>
+	</dependencies>
+
+</project>
 \ No newline at end of file
+package com.b2c.dtms.service.rpc;
+
+import com.alibaba.rocketmq.common.message.Message;
+import com.diligrp.util.exception.DiliUtilException;
+
+
+
+/**
+ * 
+ * <B>Description</B> MQ服务 <br />
+ * <B>Copyright</B> Copyright (c) 2014 www.diligrp.com All rights reserved. <br />
+ * 本软件源代码版权归地利集团,未经许可不得任意复制与传播.<br />
+ * <B>Company</B> 地利集团
+ * @createTime 2014年7月14日 下午18:15:16
+ * @author xiongdeqiang
+ */
+public interface MqService {
+	public void sendOrderMQ(Message message, String order) throws DiliUtilException;
+	
+	public void sendMQ(Message message) throws DiliUtilException;
+}
+package com.b2c.dtms.service.rpc.impl;
+
+import javax.annotation.Resource;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Service;
+
+import com.alibaba.fastjson.JSONObject;
+import com.alibaba.rocketmq.common.message.Message;
+import com.b2c.dtms.service.rpc.MqService;
+import com.diligrp.util.exception.DiliUtilException;
+import com.diligrp.util.rocketMQUtil.DiliMQProducer;
+
+@Service("rpcMqService")
+public class MqServiceImpl implements MqService {
+
+	private static final Logger LOGGER = LoggerFactory.getLogger(MqServiceImpl.class);
+
+	@Resource
+	private DiliMQProducer diliMQProducer;
+	// @Resource
+	// private ConfigurationInfoService rpcConfigInfoService;
+	// private String hotLine;
+
+	/*
+	 * @PostConstruct public void init() { hotLine =
+	 * rpcConfigInfoService.getHotLineNo(); }
+	 */
+	@Override
+	public void sendOrderMQ(Message message, String order) throws DiliUtilException {
+		LOGGER.info("发送有序MQ");
+		LOGGER.info("发送有序MQ:" + JSONObject.toJSONString(message));
+		diliMQProducer.sendOrderMsg(message, order);
+	}
+
+	@Override
+	public void sendMQ(Message message) throws DiliUtilException {
+		LOGGER.info("发送MQ");
+		diliMQProducer.sendMsg(message);
+	}
+}
+package com.b2c.dtms.handler;
+
+import java.nio.charset.Charset;
+import java.util.Map;
+
+import javax.annotation.Resource;
+
+import org.apache.commons.lang3.StringUtils;
+import org.nutz.json.Json;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Service;
+
+import com.alibaba.fastjson.JSON;
+import com.alibaba.rocketmq.common.message.Message;
+import com.b2c.dtms.common.enums.dtms.DtmsMessageTopic;
+import com.b2c.dtms.common.enums.dtms.HandleCode;
+import com.b2c.dtms.domain.DtmsMessage;
+import com.b2c.dtms.service.rpc.MqService;
+
+/**
+ * 发送MQ-消息转发专用
+ */
+@Service
+public class MqHandlerImpl extends DtmsHandler {
+	private static final Logger LOGGER = LoggerFactory.getLogger(MqHandlerImpl.class);
+	@Resource
+	private MqService rpcMqService;
+
+	@Override
+	public boolean isSupported(DtmsMessage messsage) {
+		if (messsage == null) {
+			return false;
+		}
+		return StringUtils.equals(DtmsMessageTopic.SendMq.code(), messsage.getTopic());
+	}
+
+	@SuppressWarnings("unchecked")
+	@Override
+	public boolean doExecute(DtmsHandleContext ctx) {
+		DtmsMessage dtmsMessage = ctx.getWaitHandleDtmsMessage();
+		String logMsg = null;
+		try {
+			Map<String, Object> contentMap = (Map<String, Object>) JSON.parse(dtmsMessage.getContent());
+			String topic = (String) contentMap.get("topic");
+			String tag = (String) contentMap.get("tag");
+			contentMap.remove("topic");
+			contentMap.remove("tag");
+
+			String order = String.valueOf(contentMap.get("orderId"));
+			String jsonBpdy = Json.toJson(contentMap);
+			LOGGER.info("mq body:" + jsonBpdy);
+
+			byte[] body = jsonBpdy.getBytes(Charset.forName("UTF8"));
+			Message message = new Message(topic, tag, body);
+			rpcMqService.sendOrderMQ(message, order);
+
+			logMsg = String.format("[DTMS]消息ID%s：处理成功，发送mq", dtmsMessage.getId());
+			ctx.setReturnCode(HandleCode.SUCCESS.code());
+			ctx.setReturnMessage(logMsg);
+			return true;
+		} catch (Exception e) {
+			ctx.setReturnCode(HandleCode.FAILED.code());
+			logMsg = String.format("[DTMS]消息ID%s：处理失败，发生异常[%s],需等待再次执行", dtmsMessage.getId(), e.getMessage());
+			ctx.setReturnMessage(logMsg);
+			loggerException(dtmsMessage, e);
+			return false;
+		}
+	}
+}
@@ -28,7 +28,6 @@ public class DtmsRestController {
 		DtmsMessage message = new DtmsMessage();
 		copier.copy(request, message, null);
 		message.setVersion(1L);
-		message.setTopic("CallUrl");
 		Long id;
 		try {
 			id = this.producer.produce(message);
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context"
+	xmlns:aop="http://www.springframework.org/schema/aop" xmlns:cache="http://www.springframework.org/schema/cache"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans 
+       		http://www.springframework.org/schema/beans/spring-beans.xsd 
+       		http://www.springframework.org/schema/context 
+       		http://www.springframework.org/schema/context/spring-context.xsd
+       		http://www.springframework.org/schema/aop 
+       		http://www.springframework.org/schema/aop/spring-aop.xsd
+			http://www.springframework.org/schema/cache
+			http://www.springframework.org/schema/cache/spring-cache.xsd"
+	default-autowire="byName">
+
+	<bean id="diliMQProducer" class="com.diligrp.util.rocketMQUtil.impl.DiliMQProducerImpl" />
+</beans>
 \ No newline at end of file
@@ -57,4 +57,5 @@
 	<import resource="classpath*:spring-dao.xml" />
 	<import resource="classpath*:spring-redis.xml" />
 	<import resource="classpath*:spring-dtms.xml" />
+	<import resource="classpath*:spring-rpc.xml" />
 </beans>
 \ No newline at end of file
@@ -427,7 +427,7 @@
 				<!-- 店铺接口签名 -->
 				<orders.rpc.shopClient.sign>FzZGZxd2VycXdlYXNzZHZzdnp4Y3Z</orders.rpc.shopClient.sign>
 				<!-- 消息中心配置 -->
-				<mq.namesrvAddr>10.28.11.158:9876</mq.namesrvAddr>
+				<mq.namesrvAddr>10.28.10.215:9876</mq.namesrvAddr>
 				<mq.producerGroup>orders</mq.producerGroup>
 			</properties>
 		</profile>
@@ -514,5 +514,6 @@
 		<module>dtms-dao</module>
 		<module>dtms-web</module>
 		<module>dtms-client</module>
+		<module>dtms-rpc</module>
 	</modules>
 </project>
 \ No newline at end of file