From ae661fdaf132862a8d4c1dd2bdfda3ccaaba1a5b Mon Sep 17 00:00:00 2001 From: Burce Date: Wed, 11 Sep 2019 17:46:58 +0800 Subject: [PATCH] =?UTF-8?q?[=E4=BC=98=E5=8C=96]MQ=E7=94=9F=E4=BA=A7?= =?UTF-8?q?=E5=A4=B1=E8=B4=A5=E5=9B=9E=E8=B0=83?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../src/main/resources/application-dev.yml | 1 + .../mq/impl/RabbitMqMessageProducer.java | 55 ++++++++++++------- 2 files changed, 36 insertions(+), 20 deletions(-) diff --git a/mallinkAdmin/src/main/resources/application-dev.yml b/mallinkAdmin/src/main/resources/application-dev.yml index 0ae1a7e6a..8074ce74f 100644 --- a/mallinkAdmin/src/main/resources/application-dev.yml +++ b/mallinkAdmin/src/main/resources/application-dev.yml @@ -70,6 +70,7 @@ spring: username: ENC(2f9Nqt3c4cbVGYnhpjp9Mg==) password: ENC(2av0JfMfz141IHkG6ibI8aKzmD33bc64wcN/O43ucbo=) publisher-confirms: true + publisher-returns: true virtual-host: / flyway: enabled: true diff --git a/mallinkService/src/main/java/com/iformall/mq/impl/RabbitMqMessageProducer.java b/mallinkService/src/main/java/com/iformall/mq/impl/RabbitMqMessageProducer.java index 036ed2b58..cea901409 100644 --- a/mallinkService/src/main/java/com/iformall/mq/impl/RabbitMqMessageProducer.java +++ b/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 org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.slf4j.Marker; 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.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Profile; import org.springframework.stereotype.Service; + +import javax.annotation.PostConstruct; import java.util.UUID; /** @@ -22,42 +27,52 @@ import java.util.UUID; */ @Service @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()); @Autowired - private AmqpTemplate rabbitTemplate; + private RabbitTemplate rabbitTemplate; @Autowired private WxMsgRecordMapper wxMsgRecordMapper; @Autowired 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 public void sendMessage(BaseMsg data, String topic, String tags, String keys) { + String tag = UUID.randomUUID().toString(); try { - data.setUuid(UUID.randomUUID().toString()); + data.setUuid(tag); 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()); // wxMsgRecordMapper.update(data); log.info(data.getUuid()+"发送mq成功。"); - } catch (Exception e) { - log.error("RabbitMqMessageProducer: Send Message Error ", e); + log.error(tag + " RabbitMqMessageProducer: Send Message Error", e); } } - }