文章目录
- RabbitMQ
- 入门
- 1.什么是MQ
- 2.为什么要用MQ
- 流量消峰
- 应用解耦
- 异步处理
- 3.MQ的分类
- ActiveMQ
- Kafka
- RocketMQ
- RabbitMQ
- 4.RabbitMQ
- RabbitMQ的概念
- 四大核心概念
- i.生产者 Producer
- ii.交换机 Exchange
- iii.队列 Queue
- iv.消费者 Consumer
- 核心部分
- 六大模式
- 1）简单模式 “Hello world！”
- 2）工作模式 Work queues
- 轮训分发消息
- 3）发布/订阅模式 Publish/Subscribe
- 4）路由模式 Routing
- 5）主题模式 Topics
- 6）发布确认模式 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
猜你喜欢
网友评论
- 搜索
- 最新文章
- 热门文章