一、基本概念
生产者（Producer）
产生数据发送消息的程序是生产者
交换机（Exchangs）
交换机是 RabbitMQ 非常重要的一个部件，一方面它接收来自生产者的消息，另一方面它将消息推送到队列中。交换机必须确切知道如何处理它接收到的消息，是将这些消息推送到特定队列还是推送到多个队列，亦或者是把消息丢弃，这个得有交换机类型决定
队列（Queue）
队列是 RabbitMQ 内部使用的一种数据结构，尽管消息流经 RabbitMQ 和应用程序，但它们只能存储在队列中。队列仅受主机的内存和磁盘限制的约束，本质上是一个大的消息缓冲区。许多生产者可以将消息发送到一个队列，许多消费者可以尝试从一个队列接收数据。这就是我们使用队列的方式
消费者（Consumer）
消费与接收具有相似的含义。消费者大多时候是一个等待接收消息的程序。请注意生产者，消费者和消息中间件很多时候并不在同一机器上。同一个应用程序既可以是生产者又是可以是消费者。
二、使用场景
1.电商的收货后15天自动确认收货
2.电商未付款的1小时自动取消订单
3.通过mq进行解耦合
4.通过mq缓解主进程压力
三、mq结合springboot
1.项目配置
yaml文件配置
# rabbitmq配置
rabbitmq:
host: 101.101.101.101 ## mq的ip地址
port: 5672 ## 端口号
username: cs ## 用户名
password: cs ## 密码
virtual-host: cs-cs ## 连接的库
web的访问地址为 ：101.101.101.101:15672（你的mq必须配置了mq可视化的插件）
2.mq队列的注册和绑定
1.延迟队列
@Configuration
public class delayConfig {
/**
* 延时队列交换机
*/
public static final String EXCHANG_DELAY = "exchang.delay";
/**
* 自动取消订单队列
*/
public static final String QUEUE_CLOSE = "queue.close";
/**
* 自动取消订单 绑定路由
*/
public static final String KEY_CLOSE = "key.close";
/**
* 创建延时队列交换机
*/
@Bean(EXCHANG_DELAY)
public Exchange delayExchage() {
return ExchangeBuilder
.topicExchange(EXCHANG_DELAY)
.durable(true)
.delayed() // 开启延迟消息
.build();
}
/**
* 创建订单自动关闭QUEUE
*/
@Bean(QUEUE_CLOSE)
public Queue createCloseQueue() {
return new Queue(QUEUE_CLOSE, true);
}
/**
* 创建订单自动关闭的BingdingKey
*/
@Bean(KEY_CLOSE)
public Binding bindCloseOrderKey() {
return BindingBuilder
.bind(createCloseQueue()) //绑定队列
.to(delayExchage()) //绑定延迟交换机
.with(KEY_CLOSE) //绑定路由
.noargs();
}
}
如果你想只是用这一个交换机，那你可以按照上面继续新建队列和路由，直接进行绑定就可以了，不需要构建多个延迟交换机。
2.不延迟队列
@Configuration
public class DirectConfig {
/**
* 不延迟交换机
*/
public static final String DIRECT_EXCHANGE = "direct.exchange";
/**
* 队列
*/
public static final String DIRECT_QUEUE = "direct.queue";
/**
* 路由
*/
public static final String DIRECT_KEY = "direct.key";
/**
* 创建订单延时队列交换机
*/
@Bean(DIRECT_EXCHANGE)
public Exchange directExchange() {
return ExchangeBuilder
.directExchange(DIRECT_EXCHANGE) // 队列方式
.durable(true)
.build();
}
/**
* 新建队列
*/
@Bean(DIRECT_QUEUE)
public Queue createQueue() {
return new Queue(DIRECT_QUEUE, true);
}
/**
* 队列和路由进行绑定交换机
*/
@Bean(DIRECT_KEY)
public Binding bindWithdrawKey() {
return BindingBuilder
.bind(createQueue())
.to(directExchange())
.with(DIRECT_KEY)
.noargs();
}
}
3.构建生产者
/**
* 自动确认收货（平台发货15后自动确认收货）
* 应该运营后台调用这个接口
*
* @param orderId 订单id
*/
void autoClose(Long orderId);
// 首先是serivice的实现类，现在要实现这个方法
/**
* 自动确认收货
*
* @param orderId 订单id
*/
@Override
public void autoClose(Long orderId) {
log.info("延时确认收货队列发送成功, 订单编号：{}", orderId);
// 15天
int delay = 15 * 60 * 60 * 1000;
MessagePostProcessor processor = buildProcessor(delay);//构建延迟时间
rabbitTemplate.convertAndSend( // 绑定交换机通过路由key去发送消息到对应的队列中
OrderDelayConfig.EXCHANG_DELAY,
OrderDelayConfig.KEY_CLOSE,
orderId,
processor
);
}
/**
* 构建 MessagePostProcessor
*
* @param delayTime 延时时间
*/
private MessagePostProcessor buildProcessor(int delayTime) {
return message -> {
// 设置消息持久化
message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
message.getMessageProperties().setDelay(delayTime);
return message;
};
}
///如果不需要发送和延迟消息，就不需要传这个延迟时间，其余的都是一样的
4.构建消费者
@Component
@RequiredArgsConstructor
@Slf4j
public class RabbitMQConsumer {
private final OrderService orderService;
/**
* 订单自动关闭
*
* @param orderId 订单id
*/
@RabbitListener(queues = delayConfig.QUEUE_CLOSE) //绑定队列
public void closeOrder(Long orderId) {
log.info("--------------订单:{} 自动关闭--------------", orderId);
orderService.autoDuleClose(orderId); //调用自己定义的方法进行处理
}
//这里的autoDuleClose()就是你自己的业务层处理的方法，注意：mq里面不要抛出异常；
//一定不要抛异常，出现问题需要用log.error()打印出来
}
至此就简单的实现了mq的应用，如果你需要深入了解，请自己主动的深入查询了解。
