简介
MQ全称为Message Queue, 是一种分布式应用程序的的通信方法,它是消费-生产者模型的一个典型的代表,producer往消息队列中不断写入消息,而另一端consumer则可以读取或者订阅队列中的消息。主要产品有:ActiveMQ、RocketMQ、RabbitMQ、Kafka
以rabbitMQ为例,介绍一下rabbitMQ的削峰操作。要对流量进行削峰,常用解决方案就是用消息队列来缓冲瞬时流量,把同步的直接调用转换成异步的间接推送,中间通过一个队列在一端承接瞬时的流量洪峰,在另一端平滑地将消息推送出去。
队列和消息的大小限制:
消息大小不超过4MB
队列长度限制:
消息的数量,或者是消息的总字节数,可以使用policy命令或在代码中声明来设置队列的最大长度。如果超过就有溢出操作,例如从队列的头部丢弃最老的消息。
policy命令式
定义一个消息生产者
实现方案:
定义一个消息生产者
@Test
public void test() throws Exception {
for (int i = 0; i < 1000 ; i++) {
rabbitTemplate.convertAndSend("test-queue ", "消息发送);
}
Thread.sleep(1000 * 1000);
}
使用@RabbitListener注解定义一个消息消费者
@Component
@RabbitListener(queuesToDeclare = @Queue(name = "test-queue"))
public class Consumer {
private int count = 0;
@RabbitHandler
public void receive(String msg, Channel channel, Message message) throws IOException {
long deliveryTag = message.getMessageProperties().getDeliveryTag();
try {
Thread.sleep(1000);
System.out.println("=====消息处理===>");
channel.basicAck(deliveryTag, true);
System.out.println("current count is:" + ++count);
} catch (Exception e) {
}
}
}
采用手动应答方式,消费者将消息消费以后通知RabbitMQ消息队列本条消息已经被消费,可以删除。(自动应答不管消费者对消息处理是否成功,都会告诉队列删除消息)
如何避免消息丢失:如果消费者由于某些原因失去连接(网络阻塞或消费者挂机),导致消息未发送ACK确认,RabbitMQ将了解到消息未完全处理,并将对其重新排队。如果没有此应答机制,RabbitMQ一旦向消费者传递了一条消息,便立即将该条消息标记为删除。在这种情况下,突然有个消费者挂掉了,将会丢失正在处理的消息和后续发给已挂机的消费者的消息。
spring.application.name=springboot_rabbitmq
spring.rabbitmq.host=39.107.96.73
spring.rabbitmq.port=5672
spring.rabbitmq.username=root
spring.rabbitmq.password=root
spring.rabbitmq.virtual-host=/
spring.rabbitmq.listener.simple.prefetch=2
运行效果:
参数介绍:
Ready:待消费的消息总数
Unacked:待应答的消息总数。
Total:总数 Ready+Unacked
总结:
rabbitMQ的异步、解耦和削峰三大主要功能全部提现了面向对象的思想,不仅是mq的思想精华,也是编程和架构思想的精华。尽管如此,使用mq仍然有增加系统复杂度,带来消息重复消费或消息丢失等潜在风险。具体如何权衡取舍,要看项目的实际需要,运用之妙,存乎一心。