一文带你入门消息队列RabbitMQ

x33g5p2x  于2021-12-23 转载在 RabbitMQ  
字(4.8k)|赞(0)|评价(0)|浏览(299)

📒博客首页:崇尚学技术的科班人
🏇小肖来了
🍣今天给大家带来的文章是《消息队列的入门篇》🍣
🍣我们的项目中可能会用到这个吗?我的回答是:肯定的。🍣
🍣希望各位小伙伴们能够耐心的读完这篇文章🍣
🙏博主也在学习阶段,如若发现问题,请告知,非常感谢🙏
💗同时也非常感谢各位小伙伴们的支持💗

1、消息队列

1.1、MQ的相关概念

1、什么是MQ?

  • MQ,从字面意思来看就是个队列,遵循FIFO先入先出原则,只不过队列中存放的是消息它还是一种跨进程的通信机制,用于上下游传递消息。
  • 在互联网的架构中,MQ是一种非常常见的上下游**“逻辑解耦 + 物理解耦”**的消息通信服务。使用了MQ之后,消息发送上游只需要依赖MQ,不用依赖其他服务。

2、为什么要使用MQ?

  1. 流量消峰

  • 图中的订单系统的最高限制是1万次/s。而如果我们添加了消息队列的话,它会对请求进行排队。但这在一定程度上会影响请求的速度。但总比不能下单好。
  1. 应用解耦

  • 上图的左边部分中可以看出订单系统和其他各个系统直接相连,那么如果其他各个系统中的一个出现问题的话,整个系统都会宕机。而右边部分的话,订单系统通过MQ和其他系统进行联系。当订单系统处理完自己的事情之后,由MQ对其他系统进行分配任务和监督。所以如果其中的一个系统出现问题不会影响整个系统
  1. 异步处理

  • 由上图可知,有些服务之间的调用是异步的,例如A调用B。A需要知道B什么时候将服务执行完毕。此时A就需要调用B的回调API。但是这样做,对于A来说需要每时每刻进行调用。还有一种方法就是使用MQ。当B服务结束之后,B会向MQ发送消息。然后MQ会向A发送处理消息。A服务还能即使的得到处理成功的消息

3、MQ的分类

  1. ActiveMQ
  2. kafka
  3. RocketMQ
  4. RabbitMQ

4、MQ的选择

1. kafka

  • kafka适合大量数据的互联网服务的数据收集业务。大型公司建议可以选用,如果有日志采集功能,肯定是首选kafka了。

2. RocketMQ

  • 天生为金融互联网领域而生。如果你的业务有像双11的那样的高并发的场景,建议可以选择RocketMQ

3. RabbitMQ

  • RabbitMQ结合erlang语言本身的并发优势,性能好时效性微秒级,社区活跃度也比较高,如果你的数据量没有那么大的话可以使用它。

1.2、RabbitMQ

1、RabbitMQ的概念

  • RabbitMQ是一个消息中间件;它接受并转发消息。它可以接收、存储和转发消息数据

2、四大核心概念

  1. 生产者
  2. 交换机
  3. 队列
  4. 消费者

3、RabbitMQ核心部分

4、各个名词的介绍

  • Broker:接受和分发消息的应用,也就是消息队列。
  • Virtual Host:每一个消息队列都可以划分出多个vhost,每个用户在自己的vhost创建exchange/queue等。
  • connection:生产者或消费者和消息队列之间的TCP连接。
  • channel:消息信道,channel作为轻量级的connection极大减少了操作系统建立TCP连接的开销。
  • exchange:交换机,消息到达消息队列的第一个地方,它会根据分发规则,匹配查询表中的路由关键字,分发消息到队列中。
  • queue:消息最终被送到这里等待被消费者取走。
  • binding:交换机和队列之间的虚拟连接,绑定中可以包含如有关键字,绑定信息被保存到交换机的查询表中,用于消息的分发依据。

5、安装

  1. 从官网进行下载相应的安装文件
    官网下载地址

  2. 将相关的安装文件上传到相应的/opt目录下

  1. 安装文件(按照以下顺序执行)
rpm -ivh erlang-22.3-1.el7.x86_64.rpm

yum install socat -y

rpm -ivh rabbitmq-server-3.8.8-1.el7.noarch.rpm
  1. 常用命令(按照以下顺序执行)
# 添加开机启动RabbitMQ服务
chkconfig rabbitmq-server on

# 启动服务
/sbin/service rabbitmq-server start

# 查看服务状态
/sbin/service rabbitmq-server status

# 停止服务(选择执行)
/sbin/service rabbitmq-server stop
  1. 安装并启动成功后的图片说明

  1. 开启web管理插件
rabbitmq-plugins enable rabbitmq_management

  • 需要创建用户权限。
  1. 添加一个新的用户
# 创建账号
rabbitmqctl add_user admin 123

# 设置用户角色
rabbitmqctl set_user_tags admin administrator

# 设置用户权限
set_permissions [-p <vhostpath>] <user> <conf> <write> <read>
rabbitmqctl set_permissions -p "/" admin ".*" ".*" ".*"
#用户user_admin具有/vhost1这个virtual host 中的所有资源的配置、写、读权限。

# 查看当前用户和角色
rabbitmqctl list_users
  1. 登录成功图片说明

2、HelloWorld

  • “P”是我们的生产者,“C”是我们的消费者。中间的框是一个队列 - RabbitMQ代表使用者保留的消息缓冲区。

总的说明

2.1、引入的依赖

<build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <configuration>
                    <source>8</source>
                    <target>8</target>
                </configuration>
            </plugin>
        </plugins>
    </build>

    <dependencies>
        <dependency>
            <groupId>com.rabbitmq</groupId>
            <artifactId>amqp-client</artifactId>
            <version>5.8.0</version>
        </dependency>
        <dependency>
            <groupId>commons-io</groupId>
            <artifactId>commons-io</artifactId>
            <version>2.6</version>
        </dependency>
    </dependencies>

2.2、生产者(发送消息)

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;

public class Producer {
    public static final String QUEUE_NAME = "hello";

    public static void main(String[] args) throws IOException, TimeoutException {
        // 创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        // 设置IP地址
        factory.setHost("192.168.123.129");
        // 用户名
        factory.setUsername("admin");
        // 密码
        factory.setPassword("123");

        // 创建连接
        Connection connection = factory.newConnection();
        // 获取信道
        Channel channel = connection.createChannel();
        /** * 声明队列 * 1.队列名称 * 2.队列里面的消息是否持久化(磁盘)默认情况下消息存储在内存中 * 3.该队列是否提供一个消费者进行消费,就是是否进行消息共享 * 4.就是当最后一个消费者断开连接之后,该队列是否自动删除消息 * 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(StandardCharsets.UTF_8));
        System.out.println("消息发送完毕");
    }
}

插件的观察

2.3、消费者(接收消息)

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Consumer {
    public static final String QUEUE_NAME = "hello";

    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.123.129");
        factory.setUsername("admin");
        factory.setPassword("123");

        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        DeliverCallback deliverCallback = (var1, var2)->{
            System.out.println(new String(var2.getBody()));
        };

        CancelCallback cancelCallback = var1->{
            System.out.println("消费消息被中断");
        };
        /** * 接收消息 * 1.消费哪个队列 * 2.消费成功之后是否要自动应答 * 3.消费者未成功消费的回调 * 4.消费者取消消费的回调 */
        channel.basicConsume(QUEUE_NAME,true,deliverCallback,cancelCallback);
    }
}

相关文章

微信公众号

最新文章

更多