文章目录
RabbitMQ入门1.什么是MQ2.为什么要用MQ流量消峰应用解耦异步处理
3.MQ的分类ActiveMQKafkaRocketMQRabbitMQ
4.RabbitMQRabbitMQ的概念四大核心概念i.生产者 Producerii.交换机 Exchangeiii.队列 Queueiv.消费者 Consumer
核心部分六大模式1）简单模式 “Hello world！”2）工作模式 Work queues轮训分发消息
3）发布/订阅模式 Publish/Subscribe4）路由模式 Routing5）主题模式 Topics6）发布确认模式 Publisher Confirms
RabbitMQ
入门
1.什么是MQ
RabbitMQ就是一个消息中间件
2.为什么要用MQ
流量消峰
假设一个订单系统最多被访问1万次，当用户访问超过1万次就会导致订单系统宕机，此时在用户和订单系统之间添加MQ中间件，此时用户先访问MQ，然后再去访问订单系统，因此达到消峰的能力。MQ此时对访问的用户进行排队，缺点也很明显，排队导致浪费时间，访问的速度就会下降，好处就是订单系统不会宕机。
应用解耦
一电商应用为例，应用中有订单系统、支付系统、库存系统、物流系统。用户创建订单后，如果耦合调用支付系统、库存系统、物流系统，任何一个子系统出现故障，都会造成下单异常。此时转为消息队列的方式后（在原本基础上订单系统和其他子系统之间添加MQ），当支付系统、库存系统、物流系统，任何一个子系统出现故障，队列会监督某个出现故障的子系统，直至完成结束。所以不会出现任何一个子系统出现故障影响整个系统的运转。
异步处理
有些服务调用是异步的，假设A调用B，B需要很长时间执行，但是A需要知道B什么时候可以执行完。此时有两种方式，第一种：A每过一段时间去询问B是否已经执行完。第二种：A调用B（调用完成之后A可以做其他事情），B执行完之后将消息发送给MQ，MQ在将消息发送给A表示B已经执行完成，A就收到了B执行完成的结果。
3.MQ的分类
ActiveMQ
优点：可用性高，很难丢失消息，丢失消息概率低
缺点：官方效率在5.X版本后维护越来越少，高吞吐量场景很少使用
Kafka
大数据采用、大型公司建议使用，如果有日志采集功能，首选Kafka
优点：吞吐量高、分布式，一个数据有多个副本，少数机器宕机后不会导致数据丢失
缺点：kafka 单机超过64个队列/分区，Load（加载）会发生明显的CPU飙高，队列越多，load越高，发送消息响应时间边长、消息失败不支持重试、支持消息顺序，但是一台机器宕机后就护发生消息乱序，社区更新慢。
RocketMQ
Java语言实现、天生为金融互联网领域而生，对于可靠性要求很高的场景，尤其是订单扣款，业务消峰等
优点：可用性高、分布式架构、消息做到0丢失
缺点：支持客户端语言不多，目前是Java和C++、某些系统迁移需要改动大量代码
RabbitMQ
2007年发布是一个在AMQP（高级消息队列协议）基础上完成，当前最主流的消息中间件之一。
优点：性能号、吞吐量达到万级，MQ功能比较完备、支持多种语言、支持AJAX文档齐全
缺点：商业版需要收费，学习成本较高
4.RabbitMQ
RabbitMQ的概念
RabbitMQ 是一个消息中间件：它接收/转发消息（还能存储消息）。相当于一个快递站，当你要发送一个消息时，你把你的消息放到RabbitMQ里面，然后RabbitMQ再把消息发送出去。
四大核心概念
i.生产者 Producer
谁发送消息 谁就是生产者
ii.交换机 Exchange
MQ中包含交换机在队列
一个交换机对应多个队列 （1：n关系）
交换机—绑定关系—>队列1
交换机—绑定关系—>队列2
交换机—绑定关系—>队列3
iii.队列 Queue
一个队列对应一个消费者（1：1关系）
iv.消费者 Consumer
谁接收到消息 谁就是消费者
核心部分
六大模式
1）简单模式 “Hello world！”
生产者—生产消息—>队列—接收消息—>消费者
生产者代码
/**
* 生产者：发消息
*/
public class Producer {
//队列名称
public static final String QUEUE_NAME = "hello";
//发消息
public static void main(String[] args) throws Exception {
//创建连接工厂
ConnectionFactory factory = new ConnectionFactory();
//设置工厂Ip 连接RabbitMQ队列
factory.setHost("192.168.40.128");
//设置用户名
factory.setUsername("admin");
//设置密码
factory.setPassword("1234");
//创建连接
Connection connection = factory.newConnection();
//创建信道（发送消息）
Channel channel = connection.createChannel();
/**
* 生成一个队列
* 参数1：队列名称
* 参数2：队列中的消息是否持久化
* 默认情况:消息存储在内存中.即:默认不支持持久化（持久化：保存到磁盘上）
* 参数3：该队列是否供一个消费者进行消费 是否进行消费的共享
* true：多个消费者可以消费 false：只能一个消费者消费
* 参数4：是否自动删除
* 最后一个消费者断开连接之后，该队列是否自动删除
* true：自动删除
* 参数5：其他参数
*
*/
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
//发消息
String message = "hello world";
/**
* 发送一个消息
* 参数1：发送到哪个交换机
* 参数2：路由的key值是哪个 本次是队列名称
* 参数3：其他参数信息
* 参数4：发送消息的消息体
*/
channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
System.out.println("消息发送完毕");
}
}
消费者代码
/**
* 消费者：接收消息（消费消息）
*/
public class Consumer {
public static final String QUEUE_NAME = "hello";
public static void main(String[] args) throws Exception {
//创建连接工厂
ConnectionFactory factory = new ConnectionFactory();
//设置工厂Ip
factory.setHost("192.168.40.128");
//设置用户名
factory.setUsername("admin");
//设置密码
factory.setPassword("1234");
//创建连接
Connection connection = factory.newConnection();
//创建信道（接收消息）
Channel channel = connection.createChannel();
//声明 接收消息
DeliverCallback deliverCallback = (consumerTag, message) -> {
System.out.println(new String(message.getBody()));
};
//取消消息的回调
CancelCallback cancelCallback = consumerTag -> {
System.out.println("消息接收中断");
};
/**
* 消费者接收消息
* 参数1：接收哪个队列
* 参数2：接收成功后是否要自动应答 true：自动 false：手动
* 参数3：消费者未成功接收的回调
* 参数4：消费者取消接收的回调
*/
channel.basicConsume(QUEUE_NAME, true, deliverCallback, cancelCallback);
}
}
2）工作模式 Work queues
不能重复处理同一个消息。即：一个消息最多被处理一次
生产者发送大量消息到队列中，队列发送给多个工作线程来处理消息
每个工作线程之间是竞争的关系
轮训分发消息
按周期进行
创建通用工具类
public class RabbitMqUtils {
public static Channel getChannel() throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("192.168.40.128");
factory.setUsername("admin");
factory.setPassword("1234");
Connection connection = factory.newConnection();
return connection.createChannel();
}
}
生产者：
public class Producer01 {
public static final String QUEUE_NAME = "hello";
//发送大量消息
public static void main(String[] args) throws Exception {
Channel channel = RabbitMqUtils.getChannel();
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
//从控制台输入
Scanner sc = new Scanner(System.in);
while (sc.hasNext()) {
String message = sc.next();
channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
System.out.println("发送消息完成:" + message);
}
}
}
工作线程：多个工作线程代码一致只需更换C1即可
/**
* 工作线程 == 消费者
*/
public class Worker01 {
//队列名称
public static final String QUEUE_NAME = "hello";
//接收消息
public static void main(String[] args) throws Exception {
Channel channel = RabbitMqUtils.getChannel();
//消息的接收
DeliverCallback deliverCallback = (consumerTag, message) -> {
System.out.println("接收到的消息" + new String(message.getBody()));
};
//消息接收被取消时，执行下面的内容
CancelCallback cancelCallback = consumerTag -> {
System.out.println(consumerTag + "消费者取消消费者接口回调逻辑");
};
System.out.println("C1等待接收消息......");
//消息的接收
channel.basicConsume(QUEUE_NAME, true, deliverCallback, cancelCallback);
}
}
3）发布/订阅模式 Publish/Subscribe
4）路由模式 Routing
5）主题模式 Topics
6）发布确认模式 Publisher Confirms
相关阅读
您阅读本篇文章共花了：
发表评论