Просмотр исходного кода

[优化]MQ生产失败回调

release_toaliyun_real
Burce 6 лет назад
Родитель
Сommit
ae661fdaf1
2 измененных файлов: 36 добавлений и 20 удалений
  1. +1
    -0
      mallinkAdmin/src/main/resources/application-dev.yml
  2. +35
    -20
      mallinkService/src/main/java/com/iformall/mq/impl/RabbitMqMessageProducer.java

+ 1
- 0
mallinkAdmin/src/main/resources/application-dev.yml Просмотреть файл

@@ -70,6 +70,7 @@ spring:
username: ENC(2f9Nqt3c4cbVGYnhpjp9Mg==) username: ENC(2f9Nqt3c4cbVGYnhpjp9Mg==)
password: ENC(2av0JfMfz141IHkG6ibI8aKzmD33bc64wcN/O43ucbo=) password: ENC(2av0JfMfz141IHkG6ibI8aKzmD33bc64wcN/O43ucbo=)
publisher-confirms: true publisher-confirms: true
publisher-returns: true
virtual-host: / virtual-host: /
flyway: flyway:
enabled: true enabled: true


+ 35
- 20
mallinkService/src/main/java/com/iformall/mq/impl/RabbitMqMessageProducer.java Просмотреть файл

@@ -8,11 +8,16 @@ import com.iformall.service.WxMsgRecordService;
import com.iformall.utils.JsonUtil; import com.iformall.utils.JsonUtil;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import org.slf4j.Marker;
import org.springframework.amqp.core.AmqpTemplate; import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.CorrelationData; import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Profile; import org.springframework.context.annotation.Profile;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.UUID; import java.util.UUID;


/** /**
@@ -22,42 +27,52 @@ import java.util.UUID;
*/ */
@Service @Service
@Profile(MQConfig.Impl.RABBIT_MQ) @Profile(MQConfig.Impl.RABBIT_MQ)
public class RabbitMqMessageProducer implements MqBaseProducer {
public class RabbitMqMessageProducer implements MqBaseProducer , RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {
private final Logger log = LoggerFactory.getLogger(this.getClass()); private final Logger log = LoggerFactory.getLogger(this.getClass());
@Autowired @Autowired
private AmqpTemplate rabbitTemplate;
private RabbitTemplate rabbitTemplate;
@Autowired @Autowired
private WxMsgRecordMapper wxMsgRecordMapper; private WxMsgRecordMapper wxMsgRecordMapper;
@Autowired @Autowired
private WxMsgRecordService wxMsgRecordService; private WxMsgRecordService wxMsgRecordService;
//
// @Override
// public void confirm(CorrelationData correlationData, boolean b, String s) {
// log.info("------correlationData"+correlationData.toString());
// System.out.println();
// if(b){
// System.out.println();
// }else{
//
// }
// }

@PostConstruct
public void init() {
rabbitTemplate.setConfirmCallback(this);
rabbitTemplate.setReturnCallback(this);
}

@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
log.info("confirm:: correlationData {},ack {}, cause {}", correlationData,ack,cause);
System.out.println();
if(ack){
System.out.println();
}else{

}
}


@Override
public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
log.info("returnedMessage::message {} 消息发送失败,应答码:{}, 应答描述:{} ,exchange: {} ,routingKey :{}", new String(message.getBody()), replyCode, replyText, exchange, routingKey);
}


@Override @Override
public void sendMessage(BaseMsg data, String topic, String tags, String keys) { public void sendMessage(BaseMsg data, String topic, String tags, String keys) {
String tag = UUID.randomUUID().toString();
try { try {
data.setUuid(UUID.randomUUID().toString());
data.setUuid(tag);
wxMsgRecordService.save(data); wxMsgRecordService.save(data);


//CorrelationData correlationData = new CorrelationData("16746324");
rabbitTemplate.convertAndSend(topic, JsonUtil.obj2Json(data));

CorrelationData correlationData = new CorrelationData(tag);
rabbitTemplate.convertAndSend(topic, (Object) JsonUtil.obj2Json(data),correlationData);
// data.setMsgStatus(EnumMsgRecordStatus.SEND_SUCC.getCode()); // data.setMsgStatus(EnumMsgRecordStatus.SEND_SUCC.getCode());
// wxMsgRecordMapper.update(data); // wxMsgRecordMapper.update(data);
log.info(data.getUuid()+"发送mq成功。"); log.info(data.getUuid()+"发送mq成功。");

} catch (Exception e) { } catch (Exception e) {
log.error("RabbitMqMessageProducer: Send Message Error ", e);
log.error(tag + " RabbitMqMessageProducer: Send Message Error", e);
} }
} }

} }

Загрузка…
Отмена
Сохранить