Skip to content

O
order-group
Commit 5aed9206 by ping.wu
Options

Merge branch 'feature/20221021_小程序外卖单取消订单同步取消配送单_wuping' of… 

Merge branch 'feature/20221021_小程序外卖单取消订单同步取消配送单_wuping' of https://gitlab.freemud.com/order-group-application/order-group
parents 1e210723 5894c2c8
Showing with 116 additions and 127 deletions
order-application-service/pom.xml
......@@ -374,16 +374,16 @@
<!-- <artifactId>assortment-payment-sdk</artifactId>-->
<!-- <version>2.6.10.RELEASE</version>-->
<!-- </dependency>-->
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot-starter</artifactId>
<version>2.0.4</version>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.51</version>
</dependency>
<!-- <dependency>-->
<!-- <groupId>org.apache.rocketmq</groupId>-->
<!-- <artifactId>rocketmq-spring-boot-starter</artifactId>-->
<!-- <version>2.0.4</version>-->
<!-- </dependency>-->
<!-- <dependency>-->
<!-- <groupId>com.alibaba</groupId>-->
<!-- <artifactId>fastjson</artifactId>-->
<!-- <version>1.2.51</version>-->
<!-- </dependency>-->
<dependency>
<groupId>tk.mybatis</groupId>
<artifactId>mapper-spring-boot-starter</artifactId>
......
order-application-service/src/main/java/cn/freemud/amp/config/RocketMQConfig.java
package cn.freemud.amp.config;
import cn.freemud.constant.RocketMQConst;
import com.alibaba.fastjson.support.spring.messaging.MappingFastJsonMessageConverter;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.selector.SelectMessageQueueByHash;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
import javax.annotation.Resource;
@Configuration
public class RocketMQConfig {
@Resource
private RocketMQProperties rocketMQProperties;
@Bean(name = "deliveryRocketMqTemplate", destroyMethod = "destroy")
public RocketMQTemplate rocketMQTemplate() {
DefaultMQProducer defaultMQProducer = createMQProducer(RocketMQConst.ORDER_THIRD_PUSH_PRODUCER_GROUP);
RocketMQTemplate rocketMQTemplate = new RocketMQTemplate();
rocketMQTemplate.setProducer(defaultMQProducer);
rocketMQTemplate.setMessageConverter(new MappingFastJsonMessageConverter());
rocketMQTemplate.setMessageQueueSelector(new SelectMessageQueueByHash()); // MessageQueueSelector, default SelectMessageQueueByHash
return rocketMQTemplate;
}
/**
* @return
*/
public DefaultMQProducer createMQProducer(String producerGroupName) {
RocketMQProperties.Producer producerConfig = rocketMQProperties.getProducer();
String nameServer = rocketMQProperties.getNameServer();
String groupName = StringUtils.isEmpty(producerGroupName) ? producerConfig.getGroup() : producerGroupName;
Assert.hasText(nameServer, "rocketmq.name-server must not be null");
Assert.hasText(groupName, "rocketmq.producer.group must not be null");
String accessChannel = rocketMQProperties.getAccessChannel();
String accessKey = rocketMQProperties.getProducer().getAccessKey();
String secretKey = rocketMQProperties.getProducer().getSecretKey();
DefaultMQProducer producer = null;
// 密码模式
if (!StringUtils.isEmpty(accessKey) && !StringUtils.isEmpty(secretKey)) {
producer = new DefaultMQProducer(groupName, new AclClientRPCHook(new SessionCredentials(accessKey, secretKey))
, rocketMQProperties.getProducer().isEnableMsgTrace()
, rocketMQProperties.getProducer().getCustomizedTraceTopic());
producer.setVipChannelEnabled(false);
} else {
producer = new DefaultMQProducer(groupName
, rocketMQProperties.getProducer().isEnableMsgTrace()
, rocketMQProperties.getProducer().getCustomizedTraceTopic());
}
producer.setNamesrvAddr(nameServer);
if (!StringUtils.isEmpty(accessChannel)) {
producer.setAccessChannel(AccessChannel.valueOf(accessChannel));
}
producer.setSendMsgTimeout(producerConfig.getSendMessageTimeout());
producer.setRetryTimesWhenSendFailed(producerConfig.getRetryTimesWhenSendFailed());
producer.setRetryTimesWhenSendAsyncFailed(producerConfig.getRetryTimesWhenSendAsyncFailed());
producer.setMaxMessageSize(producerConfig.getMaxMessageSize());
producer.setCompressMsgBodyOverHowmuch(producerConfig.getCompressMessageBodyThreshold());
producer.setRetryAnotherBrokerWhenNotStoreOK(producerConfig.isRetryNextServer());
producer.setPollNameServerInterval(rocketMQProperties.getPollNameServerInteval());
producer.setHeartbeatBrokerInterval(rocketMQProperties.getHeartbeatBrokerInterval());
producer.setPersistConsumerOffsetInterval(rocketMQProperties.getPersistConsumerOffsetInterval());
return producer;
}
}
//package cn.freemud.amp.config;
//
//import cn.freemud.constant.RocketMQConst;
//import com.alibaba.fastjson.support.spring.messaging.MappingFastJsonMessageConverter;
//import org.apache.rocketmq.acl.common.AclClientRPCHook;
//import org.apache.rocketmq.acl.common.SessionCredentials;
//import org.apache.rocketmq.client.AccessChannel;
//import org.apache.rocketmq.client.producer.DefaultMQProducer;
//import org.apache.rocketmq.client.producer.selector.SelectMessageQueueByHash;
//import org.apache.rocketmq.spring.core.RocketMQTemplate;
//import org.springframework.context.annotation.Bean;
//import org.springframework.context.annotation.Configuration;
//import org.springframework.util.Assert;
//import org.springframework.util.StringUtils;
//
//import javax.annotation.Resource;
//
//@Configuration
//public class RocketMQConfig {
//
// @Resource
// private RocketMQProperties rocketMQProperties;
//
// @Bean(name = "deliveryRocketMqTemplate", destroyMethod = "destroy")
// public RocketMQTemplate rocketMQTemplate() {
// DefaultMQProducer defaultMQProducer = createMQProducer(RocketMQConst.ORDER_THIRD_PUSH_PRODUCER_GROUP);
// RocketMQTemplate rocketMQTemplate = new RocketMQTemplate();
// rocketMQTemplate.setProducer(defaultMQProducer);
// rocketMQTemplate.setMessageConverter(new MappingFastJsonMessageConverter());
// rocketMQTemplate.setMessageQueueSelector(new SelectMessageQueueByHash()); // MessageQueueSelector, default SelectMessageQueueByHash
// return rocketMQTemplate;
// }
//
// /**
// * @return
// */
// public DefaultMQProducer createMQProducer(String producerGroupName) {
// RocketMQProperties.Producer producerConfig = rocketMQProperties.getProducer();
// String nameServer = rocketMQProperties.getNameServer();
// String groupName = StringUtils.isEmpty(producerGroupName) ? producerConfig.getGroup() : producerGroupName;
// Assert.hasText(nameServer, "rocketmq.name-server must not be null");
// Assert.hasText(groupName, "rocketmq.producer.group must not be null");
//
// String accessChannel = rocketMQProperties.getAccessChannel();
// String accessKey = rocketMQProperties.getProducer().getAccessKey();
// String secretKey = rocketMQProperties.getProducer().getSecretKey();
//
// DefaultMQProducer producer = null;
// // 密码模式
// if (!StringUtils.isEmpty(accessKey) && !StringUtils.isEmpty(secretKey)) {
// producer = new DefaultMQProducer(groupName, new AclClientRPCHook(new SessionCredentials(accessKey, secretKey))
// , rocketMQProperties.getProducer().isEnableMsgTrace()
// , rocketMQProperties.getProducer().getCustomizedTraceTopic());
// producer.setVipChannelEnabled(false);
// } else {
// producer = new DefaultMQProducer(groupName
// , rocketMQProperties.getProducer().isEnableMsgTrace()
// , rocketMQProperties.getProducer().getCustomizedTraceTopic());
// }
//
// producer.setNamesrvAddr(nameServer);
//
// if (!StringUtils.isEmpty(accessChannel)) {
// producer.setAccessChannel(AccessChannel.valueOf(accessChannel));
// }
//
// producer.setSendMsgTimeout(producerConfig.getSendMessageTimeout());
// producer.setRetryTimesWhenSendFailed(producerConfig.getRetryTimesWhenSendFailed());
// producer.setRetryTimesWhenSendAsyncFailed(producerConfig.getRetryTimesWhenSendAsyncFailed());
// producer.setMaxMessageSize(producerConfig.getMaxMessageSize());
// producer.setCompressMsgBodyOverHowmuch(producerConfig.getCompressMessageBodyThreshold());
// producer.setRetryAnotherBrokerWhenNotStoreOK(producerConfig.isRetryNextServer());
// producer.setPollNameServerInterval(rocketMQProperties.getPollNameServerInteval());
// producer.setHeartbeatBrokerInterval(rocketMQProperties.getHeartbeatBrokerInterval());
// producer.setPersistConsumerOffsetInterval(rocketMQProperties.getPersistConsumerOffsetInterval());
//
// return producer;
// }
//
//}
order-application-service/src/main/java/cn/freemud/amp/service/ProduceMQService.java
......@@ -5,23 +5,12 @@ import cn.freemud.amqp.Header;
import cn.freemud.amqp.MQAction;
import cn.freemud.amqp.MQMessage;
import cn.freemud.amqp.MQService;
import cn.freemud.constant.RocketMQConst;
import cn.freemud.entities.dto.delivery.CallbackUrlRequestDto;
import com.alibaba.fastjson.JSON;
import com.freemud.application.sdk.api.log.ErrorLog;
import com.google.common.base.Throwables;
import org.apache.commons.lang.BooleanUtils;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
/**
* mq发送方
*/
......@@ -30,8 +19,8 @@ public class ProduceMQService {
@Autowired
private MQService mqService;
@Resource
private RocketMQTemplate deliveryRocketMqTemplate;
// @Resource
// private RocketMQTemplate deliveryRocketMqTemplate;
@Value("${rocketmq.delivery.enable:false}")
private Boolean rocketMQDeliveryEnable;
......@@ -54,26 +43,26 @@ public class ProduceMQService {
* @param deliveryRequest
* @Description 发送配送信息到rocketMq:
*/
public void sendRocketMqOfDeliveryInfo(CallbackUrlRequestDto deliveryRequest) {
if (BooleanUtils.isNotTrue(rocketMQDeliveryEnable)) {
return;
}
Header header = new Header(MQAction.UPDATE.getAction(), "delivery/callbackUrl", deliveryRequest.getOrderId(), RocketMQConst.ORDER_THIRD_PUSH_TOPIC);
MQMessage<CallbackUrlRequestDto> mqMessage = new MQMessage(header, deliveryRequest);
String destination = String.format("%s:%s", RocketMQConst.ORDER_THIRD_PUSH_TOPIC, RocketMQConst.OrderSubsidiaryTag.delivery);
Message message = MessageBuilder
.withPayload(mqMessage)
.setHeader("partnerId", deliveryRequest.getPartnerId())
.setHeader(MessageConst.PROPERTY_KEYS, deliveryRequest.getDeliveryId())
.build();
try {
SendResult sendResult = deliveryRocketMqTemplate.syncSendOrderly(destination, message, deliveryRequest.getOrderId());
// log.debug("sendDelivery of status:{}, message:{},", sendResult.toString(), JSON.toJSONString(message));
} catch (Exception e) {
ErrorLog.printErrorLog("sendDelivery.error","delivery/callbackUrl",message,e);
}
}
// public void sendRocketMqOfDeliveryInfo(CallbackUrlRequestDto deliveryRequest) {
//
// if (BooleanUtils.isNotTrue(rocketMQDeliveryEnable)) {
// return;
// }
//
// Header header = new Header(MQAction.UPDATE.getAction(), "delivery/callbackUrl", deliveryRequest.getOrderId(), RocketMQConst.ORDER_THIRD_PUSH_TOPIC);
// MQMessage<CallbackUrlRequestDto> mqMessage = new MQMessage(header, deliveryRequest);
// String destination = String.format("%s:%s", RocketMQConst.ORDER_THIRD_PUSH_TOPIC, RocketMQConst.OrderSubsidiaryTag.delivery);
// Message message = MessageBuilder
// .withPayload(mqMessage)
// .setHeader("partnerId", deliveryRequest.getPartnerId())
// .setHeader(MessageConst.PROPERTY_KEYS, deliveryRequest.getDeliveryId())
// .build();
//
// try {
// SendResult sendResult = deliveryRocketMqTemplate.syncSendOrderly(destination, message, deliveryRequest.getOrderId());
//// log.debug("sendDelivery of status:{}, message:{},", sendResult.toString(), JSON.toJSONString(message));
// } catch (Exception e) {
// ErrorLog.printErrorLog("sendDelivery.error","delivery/callbackUrl",message,e);
// }
// }
}
order-application-service/src/main/java/cn/freemud/service/delivery/ThirdDeliveryServiceImpl.java
......@@ -194,7 +194,7 @@ public class ThirdDeliveryServiceImpl implements ThirdDeliveryService {
// 订单配送状态变化时发送到MQ
if (DeliveryStatus.checkDeliveryStatueForPlatform(deliveryStatus)) {
produceMQService.sendOfDeliveryInfo(request);
produceMQService.sendRocketMqOfDeliveryInfo(request);
// produceMQService.sendRocketMqOfDeliveryInfo(request);
}
//配送订阅消息
sendMicroMessage(partnerId, request.getOrderId(), deliveryStatus);
......
order-application-service/src/test/java/cn/freemud/amp/service/ProduceMQServiceTest.java
......@@ -37,7 +37,7 @@ public class ProduceMQServiceTest {
@Test
public void send() {
CallbackUrlRequestDto callbackUrlRequestDto = buildCallbackUrlRequestDto();
produceMQService.sendRocketMqOfDeliveryInfo(callbackUrlRequestDto);
// produceMQService.sendRocketMqOfDeliveryInfo(callbackUrlRequestDto);
}
public MQMessage buildMQMessage() {
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment