@[TOC] 一，maven引入RabbitMQ

    <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency>

@[TOC] 二，添加监听类RabbitConnectionListener，RabbitChannelListener，RabbitRecoveryListener继承自ConnectionListener

1，RabbitConnectionListener.java

package com.sojoline.wms.listener;import com.rabbitmq.client.ShutdownSignalException;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.Connection;
import org.springframework.amqp.rabbit.connection.ConnectionListener;
import org.springframework.stereotype.Service;/*** 用于监听连接的创建和关闭*/
@Slf4j
@Service
public class RabbitConnectionListener implements ConnectionListener {@Overridepublic void onCreate(Connection connection) {log.info("================onCreate: {}", connection);}@Overridepublic void onClose(Connection connection) {log.info("================onClose: {}", connection);}@Overridepublic void onShutDown(ShutdownSignalException signal) {log.info("================onShutDown: {}", signal);}
}

2，RabbitChannelListener.java

package com.sojoline.wms.listener;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ShutdownSignalException;
import com.rabbitmq.client.impl.AMQImpl;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.ChannelListener;
import org.springframework.stereotype.Service;/*** 用于监听通道的创建和销毁*/
@Slf4j
@Service
public class RabbitChannelListener implements ChannelListener {@Overridepublic void onCreate(Channel channel, boolean b) {log.info("======================onCreate channel: {}, transactional: {}", channel, b);}@Overridepublic void onShutDown(ShutdownSignalException signal) {// 可根据isHardError判断是channel断开还是connection断开if (signal.isHardError()) {AMQImpl.Connection.Close close = (AMQImpl.Connection.Close) signal.getReason();log.warn("=====================Connection onShutDown replyCode: {}, methodId: {}, classId: {}, replyText: {}",close.getReplyCode(), close.getMethodId(), close.getClassId(), close.getReplyText());} else {AMQImpl.Channel.Close close = (AMQImpl.Channel.Close) signal.getReason();log.warn("=====================Channel onShutDown replyCode: {}, methodId: {}, classId: {}, replyText: {}",close.getReplyCode(), close.getMethodId(), close.getClassId(), close.getReplyText());}}
}

3，RabbitRecoveryListener.java

package com.sojoline.wms.listener;import com.rabbitmq.client.Recoverable;
import com.rabbitmq.client.RecoveryListener;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;/*** 监听自动重连的情况，这个listener没有测试出在什么场景会出现*/
@Slf4j
@Service
public class RabbitRecoveryListener implements RecoveryListener {@Overridepublic void handleRecovery(Recoverable recoverable) {log.info("================handleRecovery: {}", recoverable);}@Overridepublic void handleRecoveryStarted(Recoverable recoverable) {log.info("================handleRecoveryStarted: {}", recoverable);}
}

4，消息投递实体类

package com.sojoline.wms.entity;import com.baomidou.mybatisplus.annotation.IdType;
import com.baomidou.mybatisplus.annotation.TableId;
import com.baomidou.mybatisplus.extension.activerecord.Model;
import lombok.Data;
import lombok.EqualsAndHashCode;
import lombok.experimental.Accessors;import java.io.Serializable;
import java.time.LocalDateTime;/*** <p>* 消息投递日志* </p>** @author Bean* @since 2020-05-07*/
@Data
@EqualsAndHashCode(callSuper = false)
@Accessors(chain = true)
public class MsgLog extends Model<MsgLog> {private static final long serialVersionUID = 1L;/*** 主键*/@TableId(value = "id", type = IdType.AUTO)private Integer id;/*** 消息唯一标识*/private Integer msgId;/*** 消息体, json格式化*/private String msg;/*** 交换机*/private String exchange;/*** 路由键*/private String routingKey;/*** 状态: 0投递中 1投递成功 2投递失败 3已消费*/private Integer status;/*** 重试次数*/private Integer tryCount;/*** 下一次重试时间*/private LocalDateTime nextTryTime;/*** 消息处理结果, json格式化*/private String result;/*** 创建时间*/private LocalDateTime createTime;/*** 更新时间*/private LocalDateTime updateTime;@Overrideprotected Serializable pkVal() {return this.id;}}

5，消息投递日志类 MsgLogService

package com.sojoline.wms.service;import com.baomidou.mybatisplus.extension.service.IService;
import com.sojoline.wms.entity.MsgLog;/*** <p>* 消息投递日志 服务类* </p>** @author Bean* @since 2020-05-06*/
public interface MsgLogService extends IService<MsgLog> {}

6，实现类

package com.sojoline.wms.service.impl;import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;import com.sojoline.wms.entity.MsgLog;
import com.sojoline.wms.mapper.MsgLogMapper;
import com.sojoline.wms.service.MsgLogService;
import org.springframework.stereotype.Service;/*** <p>* 消息投递日志 服务实现类* </p>** @author Bean* @since 2020-05-06*/
@Service
public class MsgLogServiceImpl extends ServiceImpl<MsgLogMapper, MsgLog> implements MsgLogService {}

7，消息的mapper类

package com.sojoline.wms.mapper;import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.sojoline.wms.entity.MsgLog;/*** <p>* 消息投递日志 Mapper 接口* </p>** @author Bean* @since 2020-05-06*/
public interface MsgLogMapper extends BaseMapper<MsgLog> {}

8，消息状态类

package com.autumn.learn.core.constant;public interface MsgLogStatus {/*** 消息投递中*/int DELIVERING = 0;/*** 投递成功，消息成功发送到 Exchange*/int DELIVER_EXCHANGE_SUCCESS = 1;/*** 投递失败，消息未发送到 Exchange*/int DELIVER_EXCHANGE_FAIL = 2;/*** 投递失败，消息从 Exchange 路由到 Queue 失败*/int DELIVER_QUEUE_FAIL = 3;/*** 已消费*/int CONSUMED_SUCCESS = 4;
}

@[TOC]三，添加队列常量设置和配置类RabbitMqConfiguration.java创建
1，队列常量设置

package com.autumn.learn.core.constant;public interface RabbitMqConstant {/*** 消息通知队列*/String WMS_CONTRACT_QUEUE = "wms.contract.queue";String WMS_CONTRACT_EXCHANGE = "wms.contract.exchange";String WMS_CONTRACT_KEY = "wms.contract.routing.key";/*** 消息通知ttl队列*/String WMS_CONTRACT_TTL_QUEUE = "wms.contract.ttl.queue";String WMS_CONTRACT_TTL_EXCHANGE = "wms.contract.ttl.exchange";String WMS_CONTRACT_TTL_KEY = "wms.contract.ttl.routing.key";}

2，配置类RabbitMqConfiguration.java

package com.sojoline.wms.config;import com.autumn.learn.core.constant.MsgLogStatus;
import com.autumn.learn.core.constant.RabbitMqConstant;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.sojoline.wms.entity.MsgLog;
import com.sojoline.wms.listener.RabbitChannelListener;
import com.sojoline.wms.listener.RabbitConnectionListener;
import com.sojoline.wms.listener.RabbitRecoveryListener;
import com.sojoline.wms.service.MsgLogService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Slf4j
@Configuration
public class RabbitMqConfiguration {//private RabbitAutoConfiguration rabbitAutoConfiguration;@Autowiredprivate MsgLogService msgLogService;@Autowiredprivate RabbitConnectionListener rabbitConnectionListener;@Autowiredprivate RabbitChannelListener rabbitChannelListener;@Autowiredprivate RabbitRecoveryListener rabbitRecoveryListener;@Beanpublic RabbitTemplate rabbitTemplate(CachingConnectionFactory connectionFactory) {connectionFactory.addConnectionListener(rabbitConnectionListener);connectionFactory.addChannelListener(rabbitChannelListener);connectionFactory.setRecoveryListener(rabbitRecoveryListener);//((Recoverable)connectionFactory).addRecoveryListener(rabbitRecoveryListener);// connectionFactory.setConnectionTimeout();// 配置文件已经设置// connectionFactory.setPublisherConfirms(true);// 配置文件已经设置// connectionFactory.setPublisherReturns(true);RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);// 这里的 RabbitTemplate 是新创建的配置文件中设置的 template 参数不管用rabbitTemplate.setMandatory(true);rabbitTemplate.setMessageConverter(converter());// 消息是否成功发送到Exchange，每一条发出的消息都会调用ConfirmCallbackrabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {@Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {if (ack) {// 如果消息到达exchange,则confirm回调,ack=truemsgLogService.update(null, Wrappers.<MsgLog>update().lambda().eq(MsgLog::getId, correlationData.getId()).set(MsgLog::getStatus, MsgLogStatus.DELIVER_EXCHANGE_SUCCESS));log.info("消息发送到 Exchange 成功: correlationData({}), ack({}), cause({})", correlationData, ack, cause);} else {// 如果消息没有到exchange,则confirm回调,ack=falsemsgLogService.update(null, Wrappers.<MsgLog>update().lambda().eq(MsgLog::getId, correlationData.getId()).set(MsgLog::getStatus, MsgLogStatus.DELIVER_EXCHANGE_FAIL));log.info("消息发送到 Exchange 失败: correlationData({}), cause({})", correlationData, cause);}}});// 触发setReturnCallback回调必须设置mandatory=true, 否则Exchange没有找到Queue就会丢弃掉消息, 而不会触发回调// 配置文件已经设置// rabbitTemplate.setMandatory(true);// 消息是否从Exchange路由到Queue, 注意: 这是一个失败回调, 只有在消息进入exchange但没有进入queue时才会调用。rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {@Overridepublic void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {//exchange到queue成功,则不回调return//消息从Exchange路由到Queue失败,则回调return(需设置mandatory=true,否则不回回调,消息就丢了)log.info("消息丢失:exchange({}),route({}),replyCode({}),replyText({}),message:{}", exchange, routingKey, replyCode, replyText, message);}});return rabbitTemplate;}@Beanpublic MessageConverter converter() {return new Jackson2JsonMessageConverter();}/*** 队列** @return*/@Bean@Qualifier(RabbitMqConstant.WMS_CONTRACT_QUEUE)Queue projectQueue() {return new Queue(RabbitMqConstant.WMS_CONTRACT_QUEUE, true);}/*** 延迟队列（死信队列）*/@Bean@Qualifier(RabbitMqConstant.WMS_CONTRACT_TTL_QUEUE)public Queue projectTtlQueue() {return QueueBuilder.durable(RabbitMqConstant.WMS_CONTRACT_TTL_QUEUE).withArgument("x-dead-letter-exchange", RabbitMqConstant.WMS_CONTRACT_EXCHANGE)//到期后转发的交换机.withArgument("x-dead-letter-routing-key", RabbitMqConstant.WMS_CONTRACT_KEY)//到期后转发的路由键.build();}/*** 交换器** @return*/@Bean@Qualifier(RabbitMqConstant.WMS_CONTRACT_EXCHANGE)DirectExchange projectExchange() {return (DirectExchange) ExchangeBuilder.directExchange(RabbitMqConstant.WMS_CONTRACT_EXCHANGE).durable(true).build();}/*** 延迟队列所绑定的交换机*/@Bean@Qualifier(RabbitMqConstant.WMS_CONTRACT_TTL_EXCHANGE)DirectExchange projectTtlDirect() {return (DirectExchange) ExchangeBuilder.directExchange(RabbitMqConstant.WMS_CONTRACT_TTL_EXCHANGE).durable(true).build();}/*** 声明绑定关系* 将队列绑定到交换机** @return*/@BeanBinding projectBinding(@Qualifier(RabbitMqConstant.WMS_CONTRACT_EXCHANGE) DirectExchange exchange,@Qualifier(RabbitMqConstant.WMS_CONTRACT_QUEUE) Queue queue) {return BindingBuilder.bind(queue).to(exchange).with(RabbitMqConstant.WMS_CONTRACT_KEY);}/*** 将订迟队列绑定到交换机*/@BeanBinding projectTtlBinding(@Qualifier(RabbitMqConstant.WMS_CONTRACT_TTL_EXCHANGE) DirectExchange exchange,@Qualifier(RabbitMqConstant.WMS_CONTRACT_TTL_QUEUE) Queue queue) {return BindingBuilder.bind(queue).to(exchange).with(RabbitMqConstant.WMS_CONTRACT_TTL_KEY);}}

@[TOC]四，创建消息生产者服务
1，ContractSenderService.java

package com.sojoline.wms.service;import com.sojoline.wms.vo.ContractMessageVo;public interface ContractSenderService {/*** 发送消息方法调用: 构建自定义对象消息** @param projectMessageVo* @param delayTimes* @throws Exception*/void sendProject(ContractMessageVo projectMessageVo, long delayTimes) throws Exception;
}

2，实现类ContractSenderServiceImpl.java

package com.sojoline.wms.service.impl;import com.autumn.learn.core.constant.MsgLogStatus;
import com.autumn.learn.core.constant.RabbitMqConstant;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.sojoline.wms.entity.MsgLog;
import com.sojoline.wms.service.ContractSenderService;
import com.sojoline.wms.service.MsgLogService;
import com.sojoline.wms.vo.ContractMessageVo;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.AbstractJavaTypeMapper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;import java.time.LocalDateTime;@Slf4j
@Service
public class ContractSenderServiceImpl implements ContractSenderService {@Autowiredprivate MsgLogService msgLogService;@Autowiredprivate ObjectMapper objectMapper;@Autowiredprivate RabbitTemplate rabbitTemplate;@Overridepublic synchronized void sendProject(ContractMessageVo projectMessageVo, long delayTimes) throws Exception{// step1、先落库//byte[] zipContent = ZipUtil.gzip(projectMessageVo.getContent().getBytes(StandardCharsets.UTF_8));MsgLog msgLog = new MsgLog();msgLog.setMsgId(projectMessageVo.getCode());// 存储 压缩后的 Base64 字符串//msgLog.setMsg(Base64.encode(zipContent));msgLog.setMsg(projectMessageVo.getContent());msgLog.setExchange(RabbitMqConstant.WMS_CONTRACT_TTL_EXCHANGE);msgLog.setRoutingKey(RabbitMqConstant.WMS_CONTRACT_TTL_KEY);msgLog.setStatus(MsgLogStatus.DELIVERING);msgLog.setTryCount(0);// msgLog.setNextTryTime(null);msgLog.setCreateTime(LocalDateTime.now());boolean saveOk = msgLogService.save(msgLog);if (saveOk) {projectMessageVo.setId(msgLog.getId());// step2、再发送 MQrabbitTemplate.setExchange(RabbitMqConstant.WMS_CONTRACT_TTL_EXCHANGE);rabbitTemplate.setRoutingKey(RabbitMqConstant.WMS_CONTRACT_TTL_KEY);Message message = MessageBuilder.withBody(objectMapper.writeValueAsBytes(projectMessageVo)).setDeliveryMode(MessageDeliveryMode.PERSISTENT).build();//给消息设置延迟毫秒值message.getMessageProperties().setExpiration(String.valueOf(delayTimes));message.getMessageProperties().setHeader(AbstractJavaTypeMapper.DEFAULT_CONTENT_CLASSID_FIELD_NAME, MessageProperties.CONTENT_TYPE_JSON);//消息唯一IDCorrelationData correlationData = new CorrelationData(String.valueOf(msgLog.getId()));rabbitTemplate.correlationConvertAndSend(message, correlationData);}}
}

@[TOC]五，消费者类

package com.sojoline.wms.service.impl;import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.autumn.learn.core.constant.MsgLogStatus;
import com.autumn.learn.core.constant.RabbitMqConstant;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.rabbitmq.client.Channel;
import com.sojoline.wms.entity.MsgLog;
import com.sojoline.wms.feign.KingdeeWebApiService;
import com.sojoline.wms.service.MsgLogService;
import com.sojoline.wms.service.ReceiveService;
import com.sojoline.wms.service.SessionService;
import com.sojoline.wms.vo.ContractMessageVo;
import com.sojoline.wms.vo.ReceiveOrderVo;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.ResponseEntity;
import org.springframework.stereotype.Service;import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.*;@Slf4j
@Service
public class BarcodeReceiverServiceImpl {@Autowiredprivate ObjectMapper objectMapper;@Autowiredprivate SessionService sessionService;@Autowiredprivate ReceiveService receiveService;@Autowiredprivate MsgLogService msgLogService;@Autowiredprivate KingdeeWebApiService kingdeeWebApiService;@RabbitHandler@RabbitListener(queues = RabbitMqConstant.WMS_CONTRACT_TTL_QUEUE)public void barcodeConsumer(Message message, Channel channel) {MessageProperties properties = message.getMessageProperties();Long deliveryTag = properties.getDeliveryTag();try {byte[] body = message.getBody();// 消费确认channel.basicAck(deliveryTag, false);ContractMessageVo contract = objectMapper.readValue(body, ContractMessageVo.class);int msgId = contract.getId();MsgLog msgLog = msgLogService.getById(msgId);// 消费幂等性if (null == msgLog || msgLog.getStatus().equals(MsgLogStatus.CONSUMED_SUCCESS)) {log.info("重复消费, msgId: {}", msgId);// 消费确认channel.basicAck(deliveryTag, false);return;}//具体业务逻辑 根据信息处理返回boolean值判断是否成功存入msglog表然后判断消费状态// boolean saveflag = SaveContract(cookiess, accountInfo, receiveOrderVo);//判断返回值是true活falseif (saveflag) {msgLogService.update(null, Wrappers.<MsgLog>update().lambda().eq(MsgLog::getId, msgId).set(MsgLog::getStatus, MsgLogStatus.CONSUMED_SUCCESS));// 消费确认channel.basicAck(deliveryTag, false);} else {channel.basicNack(deliveryTag, false, false);}log.info("监听消费用户日志 监听到消息： {} ", accountInfo);System.out.println(contract.getId());} catch (Exception e) {try {log.error(e.getMessage());//消费失败 抛弃此条消息 放入死信队列channel.basicNack(deliveryTag, false, false);} catch (IOException ex) {ex.printStackTrace();}}}
}

@[TOC]六，实现

 @PostMapping(value = "/submit", consumes = MediaType.APPLICATION_JSON_VALUE)public R setSubmitBarcode(@RequestBody String content) {try {if (StrUtil.isBlank(content)) {return R.ok(false);}JSONObject jsonObject = JSONUtil.parseObj(content);// 先压缩下字符串，去掉空格、换行String newContent = jsonObject.toString();ContractMessageVo contractMessageVo = new ContractMessageVo();contractMessageVo.setCode(HashUtil.fnvHash(newContent));contractMessageVo.setContent(newContent);// 毫秒值(10 分钟)long delayTimes = 10 * 60 * 1000;//调用生产者将消息存入rabbitmqcontractSender.sendProject(contractMessageVo, delayTimes);return R.ok(true);} catch (Exception e) {e.printStackTrace();}return R.ok(false);}