当前位置：首页 > news >正文

RabbitMQ--介绍

news 2025/8/5 10:25:32

一、引言

本篇文章将介绍RabbitMQ的使用，第一节就简单介绍一下RabbitMQ的作用。RabbitMQ的官方网站：RabbitMQ: One broker to queue them all | RabbitMQ

二、介绍

1.什么是MQ

MQ相当于是消息队列，基于生产者消费者模型建立，生产者将消息发送到这个中间件MQ，消费者从中间件MQ中取消息进行一定的操作。

MQ本质是一个队列，先进先出

1.同步通信

直接调用对方的服务，数据从一端直接到达另外一端。

2.异步通信

异步通信表示信息不会直接到达对方，而是通过一个中间人来数据进行转发。

2.MQ的作用

MQ主要有3个作用

1.削峰：当某个活动开启时，大量请求会直接打到服务器，导致服务器宕机，而使用了MQ之后，这些访问的消息都会达到MQ中，服务器再根据自身的实际情况去MQ中取消息进行消费。

2.解耦合：MQ可以将系统解耦，一个模块的崩溃不会影响到其他的模块

3.异步处理：当要在执行一个方法的时候，执行这个方法耗时比较长，而我们只需要这个方法的结果，这时候可以异步处理这个东西，让代码继续执行下去，之后再拿到这个结果即可，提升了时间效率

其他作用：

延迟通知：例如开会的前5分钟通知各个参会者，可以通过MQ的延迟队列进行

3.各MQ的区别

1.Kafka：比较大的MQ，追求高吞吐量，单机吞吐量达到了10w级，适合于日志的收集

2.RocketMQ：阿里巴巴开源的，比较支持Java开发，C++不成熟，适合的语言不多，入门操作比较复杂

3.Erlang语言开发，MQ功能比较完善，几乎支持所有鹅语言，吞吐量达到了万级，社区活跃度较高，比较适合小型项目

三、安装

这里介绍安装在linux操作系统下的Ubunto环境上面。

#更新软件包
sudo apt-get update#安装erlang
sudo apt-get install erlang#安装RabbitMQ
apt-get install rabbitmq-server#确认安装结果
systemctl stauts rabbitmq-server #查看rabbitmq的日志
cd /var/log/rabbitmq/ls#启用管理界面插件
rabbitmq-plugins enable rabbitmq_management#启动rabbitmq
sudo service rabbitmq-server start

默认端口号是15672，默认的用户名和密码都是guest（3.3.0之后不让继续使用了），需创建管理员账号admin，密码admin

#添加管理员用户
rabbitmqctl add_user admin admin #添加权限
rabbitmqctl set_user_tags admin administrator

操作rabbitmq的方式

1.通过代码操作：端口 5672

2.通过界面操作：端口15672

3.通过命令操作：

四、核心概念

1.rabbitmq的工作流程

生产者/消费者模型：生产者生成一条消息之后，通过一个连接（Connection）中的若干各信道（Channel）将消息转发到Broker中，Broker中有许多个虚拟主机（Virtual Host），虚拟主机中有若干个交换机（Exchange）和队列（Queue），交换机根据一定的规则将消息转发到队列中，消费者再根据连接中的信道拿到消息，对其进行消费，没有消费成功也会进行一定的操作

生产者：生产者生成消息，这个消息有一定的业务逻辑结构，里面会有业务的一些具体信息，rabbitmq根据这些标签进行路由，把消息发给感兴趣的消费者

消费者：从mq中取到消息，然后进行消费，消费失败是重新消费还是返回给mq，由自己的业务代码进行选择

Connection：客户端与mq的一个TCP连接，负责传输客户端和服务端所有数据和控制信息

Channel：消息的发送和接收都是基于Channel的

队列：存放消息的地方：一个队列可以发送消息给多个消费者，一个消费者也可以从多个队列中取消息。

交换机：用于将消息路由到队列中。根据一定的规则进行路由。

2.AMQP协议

AMQP：一种高级消息队列协议。生产者如何将消息发送给中间件，中间件如何将消息发送给消费者

五、手动代码编写RabbitMQ

建立连接所需要信息：ip，端口，账号，密码，虚拟主机

生产者：

public static void main(String[] args) throws IOException, TimeoutException {// 1.建立连接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("47.109.67.57");connectionFactory.setPort(5672);connectionFactory.setUsername("admin");connectionFactory.setPassword("admin");connectionFactory.setVirtualHost("bite");Connection connection = connectionFactory.newConnection();// 2.开启信道Channel channel  = connection.createChannel();// 3.声明交换机// 4.声明队列channel.queueDeclare("hello",true,false,false,null);// 5.发送消息String msg = "hello rabbitmq";channel.basicPublish("","hello",null,msg.getBytes());// 6.释放资源channel.close();connection.close();}

消费者：

public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {// 1.建立连接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("47.109.67.57");connectionFactory.setPort(5672);connectionFactory.setUsername("admin");connectionFactory.setPassword("admin");connectionFactory.setVirtualHost("bite");Connection connection = connectionFactory.newConnection();// 2.创建ChannelChannel channel = connection.createChannel();// 3.声明队列channel.queueDeclare("hello",true,false,false,null);// 4.消费消息DefaultConsumer consumer = new DefaultConsumer(channel){// consumerTag:消费者标签// envelop：一些配置信息，如队列名称，交换机名称等// properties：一些配置信息// body：消息的具体内容@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {// TODOSystem.out.println("消息："+new String(body));}};channel.basicConsume("hello",true,consumer);// 等待程序执行完成Thread.sleep(2000);// 5.释放资源channel.close();connection.close();}