SpringBoot整合Rabbitmq

1.pom.xml中引入rabbitmq jar包

<!-- mq启动器 -->
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

2.application.yaml中添加rabbitmq配置

springrabbitmq:publisher-confirm-type: correlatedhost: 192.168.1.141port: 30001username: hahapassword: 1234virtual-host: /HAHAtemplate:retry:# 开始重试enabled: true# 第一次与第二次投递尝试的时间间隔10秒initial-interval: 10000ms# 尝试的最大时间间隔不超过30秒max-interval: 30000ms# 上一次尝试时间间隔的乘数2*10000msmultiplier: 2mandatory: truepublisher-returns: true# 不使用自动确认 改成手动确认listener:type: simplesimple:#采用手动应答acknowledge-mode: manualhaha:mq:# 日志 交换机logExchange: haha.shop.log.exchange# 日志 成功队列logQueue: haha.shop.log.queue# 日志 路由键logRouteKey: haha.shop.log.route-key# 库存明细 交换机stockInfoExchange: haha.shop.stock.exchange# 库存明细 成功队列stockInfoQueue: haha.shop.stock.queue# 库存明细 路由键stockInfoRouteKey: haha.shop.stock.route-key

3.发送rabbitmq消息

@Resource
private RabbitTemplate rabbitTemplate;@Value("${haha.mq.stockInfoExchange}")private String exchange;@Value("${haha.mq.stockInfoRouteKey}")private String routingKey;@Overridepublic boolean sendStockInfoMQ(ProductStockInfoReq productStockInfoReq) {if (productStockInfoReq != null && productStockInfoReq.getStock() != null) {rabbitTemplate.convertAndSend(exchange, routingKey, JSONUtil.toJsonStr(productStockInfoReq));return true;}return false;}

4.接收rabbitmq消息

@Slf4j
@Service("stockInfoMqConsumer")
public class StockInfoMqConsumer {@Autowiredprivate IProductStockInfoService productStockInfoService;@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "${haha.mq.stockInfoQueue}", durable = "true"),exchange = @Exchange(name = "${haha.mq.stockInfoExchange}", type = ExchangeTypes.TOPIC),key = {"*.shop.stock.route-key"}))public void logRecordConsumer(Message message, String content, Channel channel) {log.info("店铺库存日志明细消费，消息内容：{}" , content);try {if (StrUtil.isNotEmpty(content)) {ProductStockInfoReq productStockInfoReq = JSONUtil.toBean(content, ProductStockInfoReq.class);if (productStockInfoReq.getStoreId() == null) {Integer storeId = productStockInfoService.getStoreIdByStoreId(productStockInfoReq.getStoreId());if (storeId != null) {productStockInfoReq.setStoreId(storeId);// 保存结果数据ProductStockInfo productStockInfo = new ProductStockInfo();BeanUtil.copyProperties(productStockInfoReq, productStockInfo);// 设置时间productStockInfo.setCreateTime(DateUtil.parse(productStockInfoReq.getCreateTimeStr(), DatePattern.NORM_DATETIME_FORMAT));productStockInfo.setBillType(productStockInfoReq.getRemark());boolean save = productStockInfoService.save(productStockInfo);log.info("店铺库存日志明细消费，保存结果：{}" , save);if (save) {//采用手动ack，一条条的消费channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);}} else {log.error("店铺库存日志明细消费，未查找到店铺：{}" , storeId);}} else {// 保存结果数据ProductStockInfo productStockInfo = new ProductStockInfo();BeanUtil.copyProperties(productStockInfoReq, productStockInfo);// 设置时间if(StrUtil.isNotEmpty(productStockInfoReq.getCreateTimeStr())){productStockInfo.setCreateTime(DateUtil.parse(productStockInfoReq.getCreateTimeStr(), DatePattern.NORM_DATETIME_FORMAT));}productStockInfo.setBillType(productStockInfoReq.getRemark());boolean save = productStockInfoService.save(productStockInfo);log.info("店铺库存日志明细消费，保存结果：{}" , save);if (save) {//采用手动ack，一条条的消费channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);}}} else {//采用手动ack，一条条的消费channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);}} catch (Exception e) {e.printStackTrace();}}
}