拼多商城整合 RabbitMQ
当用户下订单时,我们的业务系统直接与数据库通信,把订单保存到数据库中当系统流量突然激增,大量的订单压力,会拖慢业务系统和数据库系统
我们需要应对流量峰值,让流量曲线变得平缓,如下图
用拼多多商城展示如何应对流量暴增，RabbitMQ竟完美诠释
1.订单存储的解耦
为了进行流量削峰,我们引入 rabbitmq 消息队列,当购物系统产生订单后,可以把订单数据发送到消息队列;而订单消费者应用从消息队列接收订单消息,并把订单保存到数据库
用拼多多商城展示如何应对流量暴增，RabbitMQ竟完美诠释
这样，当流量激增时，大量订单会暂存在rabbitmq中,而订单消费者可以从容地从消息队列慢慢接收订单，向数据库保存。
2. 创建pdweb项目
从资料中导入项目即可
用拼多多商城展示如何应对流量暴增，RabbitMQ竟完美诠释
3. 定义生产者-发送订单
3.1 pom.xml 添加依赖
spring提供了更方便的消息队列访问接口,它对RabbitMQ的客户端API进行了封装,使用起来更加方便
用拼多多商城展示如何应对流量暴增，RabbitMQ竟完美诠释
3.2 修改 application.yml 文件
添加RabbitMQ的连接信息
用拼多多商城展示如何应对流量暴增，RabbitMQ竟完美诠释
3.3 修改主程序 RunPdAPP
在主程序中添加下面的方法创建Queue实例
当创建RabbitMQ连接和信道后,Spring的RabbitMQ工具会自动在服务器中创建队列,代码在RabbitAdmin.declareQueues() 方法中
用拼多多商城展示如何应对流量暴增，RabbitMQ竟完美诠释
3.4 修改 OrderServiceImpl
//RabbitAutoConfiguration中创建了AmpqTemplate实例
@Autowired
AmqpTemplate amqpTemplate;
//saveOrder原来的数据库访问代码全部注释,添加rabbitmq消息发送代码
public String saveOrder(PdOrder pdOrder) throws Exception {
String orderId = generateId();
pdOrder.setOrderId(orderId);
amqpTemplate.convertAndSend("orderQueue", pdOrder);
return orderId;
// String orderId = generateId();
// pdOrder.setOrderId(orderId);
//
//
// PdShipping pdShipping = pdShippingMapper.selectByPrimaryKey(pdOrder.getAddId());
// pdOrder.setShippingName(pdShipping.getReceiverName());
// pdOrder.setShippingCode(pdShipping.getReceiverAddress());
// pdOrder.setStatus(1);//
// pdOrder.setPaymentType(1);
// pdOrder.setPostFee(10D);
// pdOrder.setCreateTime(new Date());
//
// double payment = 0;
// List<ItemVO> itemVOs = selectCartItemByUseridAndItemIds(pdOrder.getUserId(), pdOrder.getItemIdList());
// for (ItemVO itemVO : itemVOs) {
// PdOrderItem pdOrderItem = new PdOrderItem();
// String id = generateId();
// //String id="2";
// pdOrderItem.setId(id);
// pdOrderItem.setOrderId(orderId);
// pdOrderItem.setItemId("" + itemVO.getPdItem().getId());
// pdOrderItem.setTitle(itemVO.getPdItem().getTitle());
// pdOrderItem.setPrice(itemVO.getPdItem().getPrice());
// pdOrderItem.setNum(itemVO.getPdCartItem().getNum());
//
// payment = payment + itemVO.getPdCartItem().getNum() * itemVO.getPdItem().getPrice();
// pdOrderItemMapper.insert(pdOrderItem);
// }
// pdOrder.setPayment(payment);
// pdOrderMapper.insert(pdOrder);
// return orderId;
}
4. 消费者-接收订单,并保存到数据库
4.1 pd-web项目复制为pd-order-consumer
用拼多多商城展示如何应对流量暴增，RabbitMQ竟完美诠释
4.2 修改 application.yml 文件
把端口修改成 81
用拼多多商城展示如何应对流量暴增，RabbitMQ竟完美诠释
4.3 新建 OrderConsumer
用拼多多商城展示如何应对流量暴增，RabbitMQ竟完美诠释
4.4 修改 OrderServiceImpl 的 saveOrder() 方法
public String saveOrder(PdOrder pdOrder) throws Exception {
// String orderId = generateId();
// pdOrder.setOrderId(orderId);
//
// amqpTemplate.convertAndSend("orderQueue", pdOrder);
// return orderId;
//
//
//
// String orderId = generateId();
// pdOrder.setOrderId(orderId);
//从RabbitMQ接收的订单数据,
//已经在上游订单业务中生成过id,这里不再重新生成id
//直接获取该订单的id
String orderId = pdOrder.getOrderId();
PdShipping pdShipping = pdShippingMapper.selectByPrimaryKey(pdOrder.getAddId());
pdOrder.setShippingName(pdShipping.getReceiverName());
pdOrder.setShippingCode(pdShipping.getReceiverAddress());
pdOrder.setStatus(1);//
pdOrder.setPaymentType(1);
pdOrder.setPostFee(10D);
pdOrder.setCreateTime(new Date());
double payment = 0;
List<ItemVO> itemVOs = selectCartItemByUseridAndItemIds(pdOrder.getUserId(), pdOrder.getItemIdList());
for (ItemVO itemVO : itemVOs) {
PdOrderItem pdOrderItem = new PdOrderItem();
String id = generateId();
//String id="2";
pdOrderItem.setId(id);
pdOrderItem.setOrderId(orderId);
pdOrderItem.setItemId("" + itemVO.getPdItem().getId());
pdOrderItem.setTitle(itemVO.getPdItem().getTitle());
pdOrderItem.setPrice(itemVO.getPdItem().getPrice());
pdOrderItem.setNum(itemVO.getPdCartItem().getNum());
payment = payment + itemVO.getPdCartItem().getNum() * itemVO.getPdItem().getPrice();
pdOrderItemMapper.insert(pdOrderItem);
}
pdOrder.setPayment(payment);
pdOrderMapper.insert(pdOrder);
return orderId;
}
5. 手动确认
5.1 修改 application.yml 文件
spring:
rabbitmq:
listener:
simple:
acknowledge-mode: manual`
5.2 OrderConsumer
package com.pd;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import com.pd.pojo.PdOrder;
import com.pd.service.OrderService;
import com.rabbitmq.client.Channel;
@Component
public class OrderConsumer {
//收到订单数据后,会调用订单的业务代码,把订单保存到数据库
@Autowired
private OrderService orderService;
//添加该注解后,会从指定的orderQueue接收消息,
//并把数据转为 PdOrder 实例传递到此方法
@RabbitListener(queues="orderQueue")
public void save(PdOrder pdOrder, Channel channel, Message message)
{
System.out.println("消费者");
System.out.println(pdOrder.toString());
try {
orderService.saveOrder(pdOrder);
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
} catch (Exception e) {
e.printStackTrace();
}
}
}
希望这篇文章对大家理解RabbitMQ并有所帮助。
如果觉得本文对你有帮助，可以转发关注支持一下。
</article>
网友评论