遇见的问题:
- 假如出现slf4j的错误我们需要再导入一个包
<dependency> <groupId>org.slf4j</groupId> <artifactId>slf4j-simple</artifactId> <version>1.7.32</version> </dependency>
- 假如出现错误如下
connection error; protocol method: #method<connection.close>(reply-code=530, reply-text=N
我们需要再次打开操作,然后对其进行set permison即可
- 我们一定要打开端口号 5672 这个端口号!!!
5672: RabbitMQ的通讯端口 25672: RabbitMQ的节点间的CLI通讯端口是 15672: RabbitMQ HTTP_API的端口,管理员用户才能访问,用于管理RabbitMQ,需要启动Management插件。 1883,8883: MQTT插件启动时的端口。 61613、61614: STOMP客户端插件启用的时候的端口。 15674、15675: 基于webscoket的STOMP端口和MOTT端口
- 消费者
package com.jsxs.rabbitmq.simple; import com.rabbitmq.client.*; import java.io.IOException; import java.nio.charset.StandardCharsets; import java.util.concurrent.TimeoutException; public class Consumer { public static void main(String[] args) throws IOException, TimeoutException { // 所有的中间件技术都是基于tcp/ip协议基础之上的协议规范,只不过rabbitmq遵循的是 ampq 协议也就是tcp/ip是基础. //port ip // 1.创建链接工厂 ConnectionFactory connectionFactory = new ConnectionFactory(); connectionFactory.setHost("8.130.48.9"); // ip connectionFactory.setPort(5672); // port connectionFactory.setUsername("admin"); //username connectionFactory.setPassword("XXXX"); // password connectionFactory.setVirtualHost("/"); // 消息放在哪 // 2.创建链接 Connection Connection connection = connectionFactory.newConnection("生产者"); // 工厂创建链接 // 3.通过链接获取通道Channel Channel channel = connection.createChannel(); // 链接创建通道 // 4.通过创建交换机,声明队列,绑定关系,路由Key,发送消息和接受消息 channel.basicConsume("queue1", true, new DeliverCallback() { @Override public void handle(String s, Delivery meassage) throws IOException { System.out.println(new String("收到消息是" + new String(meassage.getBody(),"UTF-8"))); } }, new CancelCallback() { @Override public void handle(String s) throws IOException { System.out.println("接受消息失败!!!"); } }); System.out.println("开始接受消费"); System.in.read(); //进行阻断 阻塞 // 7.关闭通道 if (channel!=null&&channel.isOpen()){ channel.close(); // 先关闭通道 connection.close(); //再关闭链接 System.out.println("消息发送成功!!!"); } } }
首先我们debug一下生产者让其过渡到MQ缓存,然后MQ缓存存放生产者的信息。然后一直等待消费者来接受。
非持久化队列: 我们消费者消除之后,那么MQ缓存中就会消除。
持久化队列:我们消费之后并不会移除队列,依然会存储。
2. 什么是AMQP
(1). 什么是AMQP
AMQP全称:Advanced Message Queuing Protocol(高级消息队列协议)。是应用层协议的一个开发标准,为面向消息的中间件设计.
(2) AMQP生产者流转过程
面试题: 为什么RabbitMQ是基于channel去处理而不是链接?
因为链接需要走的是"三次握手四次挥手"这个原理,基本效率是比较低的。所以我们使用了信道的方式对其进行运行
(3). AMQP消费者流转过程
3.RabbitMQ的核心组成部分
(1).RabbitMQ的核心组成部分
String message = "Hello jsxs!!!"; // 6: 发送消息给队列queue // @params1: 交换机 @params2 队列、路由key @params 消息的状态控制 @params4 消息主题 // 面试题:可以存在没有交换机的队列吗?不可能,虽然没有指定交换机但是一定会存在一个默认的交换机。 channel.basicPublish("", queueName, null, message.getBytes());
核心概念:
Server: 又称Broker,接受客户端的链接,实现AMQP实体服务
,安装Rabbitma-server
Connection: 连接,应用程序与Broker的网络链接 TCP/IP/协议:“三次握手和四次挥手”
Channel: 网络通信,几乎所有的操作都在Channel中进行
,Channel是进行消息读写的通道,客户端可以建立对各Channel,每个Channel代表一个会话任务
。
Message:消息,服务于应用程序之间的传送的数据,由Properties组成,Properties可以是对消息进行修饰
,比如消息的优先级,延迟等高级特性,Body则就是消息体的内容
Virtual Host: 虚拟地址,用于进行逻辑隔离,最上面的消息路由,一个虚拟主机理由可以有若干个Exchange和Queueu,同一个虚拟机里面不能有相同名字的Exchange
.
Exchange : 交换机,接受消息,根据路由键发送消息到绑定的队列
。
Bindings: Exchange和Queue之间的虚拟链接,Bingding中可以保护多个 Routinue key
。
Routing key: 是一个路由规则
,虚拟机可以用它来确定如何路由一个特定消息。
Queue : 队列也成为Message Queue,消息队列,保存消息并将他们转发给消费者
。
在正式开发中: 我们的交换机一定要指定一个交换机,尽量不要去使用默认的交换机。
(2).RabbitMQ整体架构是什么样子的?
生产者->交互机->消息队列(MQ)->消费者
(3).RabbitMQ的运行流程
(4).RabbitMQ支持的消息模型
总共六种:
https://www.rabbitmq.com/getstarted.html
一、 简单模式
功能:一个生产者P发送消息到队列Q,一个消费者C接收
① 生产者实现思路:
创建连接工厂ConnectionFactory,设置服务地址127.0.0.1,端口号5672,设置用户名、密码、virtual host,从连接工厂中获取连接connection,使用连接创建通道channel,使用通道channel创建队列queue,使用通道channel向队列中发送消息,关闭通道和连接。
② 消费者实现思路:
创建连接工厂ConnectionFactory,设置服务地址127.0.0.1,端口号5672,设置用户名、密码、virtual host,从连接工厂中获取连接connection,使用连接创建通道channel,使用通道channel创建队列queue, 创建消费者并监听队列,从队列中读取消息。
二、 工作模式
功能:一个生产者,多个消费者,每个消费者获取到的消息唯一,多个消费者只有一个队列
任务队列:避免立即做一个资源密集型任务,必须等待它完成,而是把这个任务安排到稍后再做。我们将任务封装为消息并将其发送给队列。后台运行的工作进程将弹出任务并最终执行作业。当有多个worker同时运行时,任务将在它们之间共享。
三、 publish/subscribe发布订阅(共享资源)
功能:一个生产者发送的消息会被多个消费者获取。一个生产者、一个交换机、多个队列、多个消费者。
生产者:可以将消息发送到队列或者是交换机。
消费者:只能从队列中获取消息。
① 生产者实现思路:
创建连接工厂ConnectionFactory,设置服务地址127.0.0.1,端口号5672,设置用户名、密码、virtual host,从连接工厂中获取连接connection,使用连接创建通道channel,使用通道channel创建队列queue,使用通道channel创建交换机并指定交换机类型为fanout,使用通道向交换机发送消息,关闭通道和连接。
② 消费者实现思路:
创建连接工厂ConnectionFactory,设置服务地址127.0.0.1,端口号5672,设置用户名、密码、virtual host,从连接工厂中获取连接connection,使用连接创建通道channel,使用通道channel创建队列queue,绑定队列到交换机,设置Qos=1,创建消费者并监听队列,使用手动方式返回完成。可以有多个队列绑定到交换机,多个消费者进行监听。
四、 路由模式
该方式一个路由键对应一个消息队列,一个消息队列可以对应多个路由键,一个消息队列对应一个消费者,当一个队列下有多个消费者时,MQ采用的是轮询机制,选取一个消费者消费该队列下的消息,其他消费者则轮空。该模式给消息指明了准确的路线,告诉消息必须按照我制定的路线规则来走,适合于比较简单的场景,缺点是路由规则不够灵活。
五、 主题模式
基本思想和路由模式是一样的,只不过路由键支持模糊匹配,符号“#”匹配一个或多个词,符号“*”匹配不多不少一个词,路由规则变得灵活多变,可拓展性非常的强
应用场景:
② 单发送,单接收的应用场景
③ 多发送,单接收的应用场景(主要)
六、 RPC ->我们不常用的模式