|
|
|
@ -2,14 +2,13 @@ package com.bnyer.img.service.impl; |
|
|
|
|
|
|
|
import com.alibaba.fastjson.JSON; |
|
|
|
import com.alibaba.fastjson.JSONObject; |
|
|
|
import com.bnyer.common.core.constant.ServiceNameConstants; |
|
|
|
import com.bnyer.common.core.domain.ImgMqMessageRecord; |
|
|
|
import com.bnyer.common.core.enums.EnumMessageStatus; |
|
|
|
import com.bnyer.common.core.utils.bean.EntityConvertUtil; |
|
|
|
import com.bnyer.common.core.utils.uuid.IdUtils; |
|
|
|
import com.bnyer.common.rocketmq.constant.RocketMqConstant; |
|
|
|
import com.bnyer.common.rocketmq.domain.img.ImgMqLocalRecordMessage; |
|
|
|
import com.bnyer.common.rocketmq.domain.img.ImgMqLocalRecordMessage; |
|
|
|
import com.bnyer.common.rocketmq.constant.RocketMqTopic; |
|
|
|
import com.bnyer.common.rocketmq.domain.MqRecordMessage; |
|
|
|
import com.bnyer.common.rocketmq.template.RocketMQEnhanceTemplate; |
|
|
|
import com.bnyer.img.mapper.ImgMqMessageRecordMapper; |
|
|
|
import com.bnyer.img.service.ImgMqMessageRecordService; |
|
|
|
@ -18,11 +17,13 @@ import org.apache.rocketmq.client.producer.SendCallback; |
|
|
|
import org.apache.rocketmq.client.producer.SendResult; |
|
|
|
import org.apache.rocketmq.spring.core.RocketMQTemplate; |
|
|
|
import org.apache.rocketmq.spring.support.RocketMQHeaders; |
|
|
|
import org.springframework.core.env.Environment; |
|
|
|
import org.springframework.messaging.Message; |
|
|
|
import org.springframework.messaging.support.MessageBuilder; |
|
|
|
import org.springframework.stereotype.Service; |
|
|
|
import org.springframework.transaction.annotation.Transactional; |
|
|
|
|
|
|
|
import javax.annotation.PostConstruct; |
|
|
|
import javax.annotation.Resource; |
|
|
|
import java.util.Date; |
|
|
|
|
|
|
|
@ -42,6 +43,16 @@ public class ImgMqMessageRecordServiceImpl implements ImgMqMessageRecordService |
|
|
|
@Resource |
|
|
|
private ImgMqMessageRecordMapper imgMqMessageRecordMapper; |
|
|
|
|
|
|
|
@Resource |
|
|
|
private Environment env; |
|
|
|
|
|
|
|
private String applicationName; |
|
|
|
|
|
|
|
@PostConstruct |
|
|
|
public void init(){ |
|
|
|
this.applicationName = env.getProperty("spring.application.name"); |
|
|
|
} |
|
|
|
|
|
|
|
/** |
|
|
|
* 发送同步步消息 |
|
|
|
* @param topic |
|
|
|
@ -59,17 +70,18 @@ public class ImgMqMessageRecordServiceImpl implements ImgMqMessageRecordService |
|
|
|
log.info("消息发送中,开始入库本地消息记录表"); |
|
|
|
ImgMqMessageRecord imgMqMessageRecord = saveMessageRecord(topic,tag, message); |
|
|
|
//发消息
|
|
|
|
ImgMqLocalRecordMessage mqLocalMessage = EntityConvertUtil.copy(imgMqMessageRecord, ImgMqLocalRecordMessage.class); |
|
|
|
mqLocalMessage.setSource(ServiceNameConstants.ORDER_SERVICE); |
|
|
|
Message<ImgMqLocalRecordMessage> sendMessage = MessageBuilder.withPayload(mqLocalMessage).setHeader(RocketMQHeaders.KEYS, imgMqMessageRecord.getMessageKey()).build(); |
|
|
|
MqRecordMessage mqRecordMessage = EntityConvertUtil.copy(imgMqMessageRecord, MqRecordMessage.class); |
|
|
|
mqRecordMessage.setSource(applicationName); |
|
|
|
mqRecordMessage.setReturnTopic(RocketMqTopic.IMG_RETURN_MSG_TOPIC); |
|
|
|
Message<MqRecordMessage> sendMessage = MessageBuilder.withPayload(mqRecordMessage).setHeader(RocketMQHeaders.KEYS, imgMqMessageRecord.getMessageKey()).build(); |
|
|
|
SendResult sendResult; |
|
|
|
try { |
|
|
|
sendResult = rocketMQTemplate.syncSend(rocketMQEnhanceTemplate.buildDestination(topic,tag), sendMessage); |
|
|
|
log.info("topic:[{}]tag:[{}]同步消息[{}]发送结果[{}]", topic,tag, JSONObject.toJSON(message), JSONObject.toJSON(sendResult)); |
|
|
|
} catch (Exception e) { |
|
|
|
editMessageRecordStatus(imgMqMessageRecord.getId(), imgMqMessageRecord.getStatus(),e.getMessage()); |
|
|
|
log.info("topic:[{}]tag:[{}]同步消息[{}]发送结果[{}]", topic,tag, JSONObject.toJSON(message), e.getMessage()); |
|
|
|
log.info("消息发送失败，更新消息记录，等待定时任务进行消息补偿"); |
|
|
|
editMessageRecordStatus(imgMqMessageRecord.getId(), imgMqMessageRecord.getStatus(),e.getMessage()); |
|
|
|
// throw new RuntimeException(e);
|
|
|
|
} |
|
|
|
} |
|
|
|
@ -86,9 +98,10 @@ public class ImgMqMessageRecordServiceImpl implements ImgMqMessageRecordService |
|
|
|
log.info("消息发送中,开始入库本地消息记录表"); |
|
|
|
ImgMqMessageRecord imgMqMessageRecord = saveMessageRecord(topic,tag, message); |
|
|
|
//发消息
|
|
|
|
ImgMqLocalRecordMessage mqLocalMessage = EntityConvertUtil.copy(imgMqMessageRecord, ImgMqLocalRecordMessage.class); |
|
|
|
mqLocalMessage.setSource(ServiceNameConstants.IMG_SERVICE); |
|
|
|
rocketMQTemplate.asyncSend(rocketMQEnhanceTemplate.buildDestination(topic,tag), MessageBuilder.withPayload(mqLocalMessage).build(), new SendCallback() { |
|
|
|
MqRecordMessage mqRecordMessage = EntityConvertUtil.copy(imgMqMessageRecord, MqRecordMessage.class); |
|
|
|
mqRecordMessage.setSource(applicationName); |
|
|
|
mqRecordMessage.setReturnTopic(RocketMqTopic.IMG_RETURN_MSG_TOPIC); |
|
|
|
rocketMQTemplate.asyncSend(rocketMQEnhanceTemplate.buildDestination(topic,tag), MessageBuilder.withPayload(mqRecordMessage).build(), new SendCallback() { |
|
|
|
@Override |
|
|
|
public void onSuccess(SendResult sendResult) { |
|
|
|
log.info("消息发送成功,topic:{},tag:{},message:{},result:{}",topic,tag,JSON.toJSONString(message), JSON.toJSONString(sendResult)); |
|
|
|
@ -97,8 +110,7 @@ public class ImgMqMessageRecordServiceImpl implements ImgMqMessageRecordService |
|
|
|
public void onException(Throwable throwable) { |
|
|
|
log.error("消息发送失败,topic:{},tag:{},message:{},error:{}",topic,tag,JSON.toJSONString(message),throwable.getMessage()); |
|
|
|
log.info("消息发送失败，更新消息记录，等待定时任务进行消息补偿"); |
|
|
|
imgMqMessageRecord.setErrMsg(throwable.getMessage()); |
|
|
|
editMessageRecordStatus(imgMqMessageRecord.getId(),imgMqMessageRecord.getStatus(),imgMqMessageRecord.getErrMsg()); |
|
|
|
editMessageRecordStatus(imgMqMessageRecord.getId(),imgMqMessageRecord.getStatus(),throwable.getMessage()); |
|
|
|
} |
|
|
|
}); |
|
|
|
} |
|
|
|
@ -115,9 +127,10 @@ public class ImgMqMessageRecordServiceImpl implements ImgMqMessageRecordService |
|
|
|
log.info("消息发送中,开始入库本地消息记录表"); |
|
|
|
ImgMqMessageRecord imgMqMessageRecord = saveMessageRecord(topic,tag, message); |
|
|
|
//发消息
|
|
|
|
ImgMqLocalRecordMessage mqLocalMessage = EntityConvertUtil.copy(imgMqMessageRecord, ImgMqLocalRecordMessage.class); |
|
|
|
mqLocalMessage.setSource(ServiceNameConstants.ORDER_SERVICE); |
|
|
|
rocketMQTemplate.asyncSend(rocketMQEnhanceTemplate.buildDestination(topic,tag), MessageBuilder.withPayload(mqLocalMessage).build(), new SendCallback() { |
|
|
|
MqRecordMessage mqRecordMessage = EntityConvertUtil.copy(imgMqMessageRecord, MqRecordMessage.class); |
|
|
|
mqRecordMessage.setSource(applicationName); |
|
|
|
mqRecordMessage.setReturnTopic(RocketMqTopic.IMG_RETURN_MSG_TOPIC); |
|
|
|
rocketMQTemplate.asyncSend(rocketMQEnhanceTemplate.buildDestination(topic,tag), MessageBuilder.withPayload(mqRecordMessage).build(), new SendCallback() { |
|
|
|
@Override |
|
|
|
public void onSuccess(SendResult sendResult) { |
|
|
|
log.info("消息发送成功,topic:{},tag:{},message:{},result:{}",topic,tag,JSON.toJSONString(message),JSON.toJSONString(sendResult)); |
|
|
|
|
