[RabbitMQ] RabbitMQ简单概述,用法和交换机模型

  • Post author:
  • Post category:其他




MQ概述:

Message Queue(消息队列),实在消息的传输过程中保存消息的容器,都用于分布式系统之间进行通信

分布式系统通信的两种方式:直接远程调用 和 借助第三昂 完成间接通信

发送方称谓生产者,接收方称为消费者



MQ优势和劣势

优势:

  1. 应用解耦:如 把一个系统的不同功能拆分出来,中间连接mq,那么其中一个功能的系统挂掉也不会影响其它系统,达到解耦
  2. 异步提速:如 一个订单接受下单请求,正常需要去库存,支付,物流三个系统走一遍,但是可以通过直接交手mq然后直接返回,那三个系统用mq去排队进行操作,大大缩短等待时间,增加系统吞吐量
  3. 削峰填谷

劣势:

  1. 系统可用性降低
  2. 系统复杂性提高
  3. 一致性问题

RabbitMQ时基于AMQP(高级信息队里额协议)用于软件通信的中间件

RabbitMQ四大核心

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

AMQP协议是一种二进制协议,它定义了一组规则和标准,以确保消息可以在不同的应用程序和平台之间传递和解释,AMQP包含四个核心组件:

  1. 消息
  2. 交换机
  3. 队列
  4. 绑定



工作原理

Broker:接受和分发消息的应用。RabbitMQ Server 就是 Message Broker

Virtual host:是一个虚拟主机的概念,一个Broker可以有多个Virtual host,每个Virtual都有一套自己的Exchange(交换机)和Queue(队列)

RabbitMQ工作原理



快速上手

  1. 导入依赖
    <dependency>
      <groupId>com.rabbitmq</groupId>
      <artifactId>amqp-client</artifactId>
      <version>5.16.0</version>
    </dependency>
  1. 写一个生产者
public class Producer {
    public static void main(String[] args) throws Exception {

        String queueName = "test_queue_name";
        String exchangeName = "test_exchange_name";
        //创建链接工厂
        ConnectionFactory factory = new ConnectionFactory();
        //服务地址
        factory.setHost("xx.xx.xx.xx");
        //账号密码
        factory.setUsername("root");
        factory.setPassword("xxx");
        //端口号
        factory.setPort(5672);
        //创建链接
        Connection connection = factory.newConnection();
        //创建信道
        Channel channel = connection.createChannel();

        /**
         * 创建交换机
         * params:
         * 1. 交换机名称
         * 2. 交换机类型:direct,topic,fanout,headers
         * 3. 指定交换机是否持久化 - true:那么交换机的元数据要持久化
         * 4. 指定交换机没有队列绑定时是否需要删除
         * 5. Map<String,Object>用来指定我们交换机其他的结构化参数,
         */
        channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT,true,false,null);

        /**
         * 生成一个队列
         * params
         * 1. 队列名称
         * 2. 队列是否需要持久化 - 是队列名称等这些元数据的持久化,不是队列中消息的持久化
         * 3. 标识队列是不是私有的,如果是私有的,只有创建它的应用程序才能消费消息
         * 4. 队列在没有消费者订阅的情况下是否自动删除
         * 5. 队列的一些结构化信息,比如声明私信队列,磁盘队列会用到
         */
        channel.queueDeclare(queueName,true,false,false,null);

        /**
         * 将交换机和队列绑定
         * params:
         * 1.队列名称
         * 2. 交换机名称
         * 3. 路由键 - 在直连模式下,可以为我们队列名称
         */
        channel.queueBind(queueName,exchangeName,queueName);

        /**发送消息
         * params:
         * 1. 发送到哪个交换机
         * 2. 队列名称
         * 3. 其他参数信息
         * 4. 发送消息的消息体
         */
        String msg = "hello RabbitMQ!";
        channel.basicPublish(exchangeName,queueName,null,msg.getBytes());
        channel.close();
        connection.close();
    }
}
  1. 再写一个消费者来消费消息
public class Consumer {
    public static void main(String[] args) throws Exception{
        String queueName = "test_queue_name";
        String exchangeName = "test_exchange_name";
        //创建链接工厂
        ConnectionFactory factory = new ConnectionFactory();
        //服务地址
        factory.setHost("xx.xx.xx.xx");
        //账号密码
        factory.setUsername("root");
        factory.setPassword("xxx");
        //端口号
        factory.setPort(5672);
        //创建链接
        Connection connection = factory.newConnection();
        //创建信道
        Channel channel = connection.createChannel();

        //接收消息的回调函数
        DeliverCallback deliverCallback = (consumerTage, message) -> {
            System.out.println("接收到消息" + new String(message.getBody()));
        };

        //取消消息的回调函数
        CancelCallback cancelCallback = consumerTage -> {
            System.out.println("消费消息被中断");
        };
        /**
         * 消费消息
         * Params:
         * 1. 消费那个队列
         * 2. 消费成功后是否自动应答
         * 3. 接收消息的回调函数
         * 4. 取消消息的回调函数
         */
        channel.basicConsume(queueName,true,deliverCallback,cancelCallback);
    }
}



RabbitMQ的交换机类型

  1. direct

    路由键与队列名完全匹配的交换机,通过RoutingKey路由键将交换机和队列进行绑定
public class Producer {
    public static void main(String[] args) throws Exception {



        String exchangeName = "test_exchange_name";
        String queueName1 = "test_queue_name1";
        String queueName2 = "test_queue_name2";
        String queueName3 = "test_queue_name3";
        String queueName4 = "test_queue_name4";

        String key1 = "key_1";
        String key3 = "key_3";
        String key4 = "key_4";

        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("xx.xxx.xxx.xx");
        factory.setUsername("root");
        factory.setPassword("xxx");
        factory.setPort(5672);
        Connection connection = factory.newConnection();       
        Channel channel = connection.createChannel();

        channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT,true,false,null);

        channel.queueDeclare(queueName1,true,false,false,null);
        channel.queueDeclare(queueName2,true,false,false,null);
        channel.queueDeclare(queueName3,true,false,false,null);
        channel.queueDeclare(queueName4,true,false,false,null);

        channel.queueBind(queueName1,exchangeName,key1);
        channel.queueBind(queueName2,exchangeName,key1);
        channel.queueBind(queueName3,exchangeName,key3);
        channel.queueBind(queueName4,exchangeName,key4);

        channel.basicPublish(exchangeName,key1,null,"hello Key1".getBytes());
        channel.basicPublish(exchangeName,key3,null,"hello Key3".getBytes());
        channel.basicPublish(exchangeName,key4,null,"hello Key4".getBytes());

        channel.close();
        connection.close();
        System.out.println("发送消息成功");
    }
}

如上述代码,执行过后,四个队列每个都会收到消息,其中test_queue_name1,test_queue_name2各收到了一条hello Key1

direct的拼配要求是精准的,设置路由为key1就只能向key1发送才会接收到,发送key或者key.1之类都是不会接收到的

2. fanout

扇出类型交换机,会将消息分发给所有绑定了此交换机的队列,此时RoutingKey参数无效

也就是说,fanout类型交换机下发一条消息,无论RoutingKey是什么,所有队列都会收到消息

  1. topic

    主题类型交换机,此种交换机与Direct类似,也需要routingkey进行匹配分发,区别在于可以模糊查询

    Topic中,将routingkey通过”.”来分为多个部分

    “*”:代表一个部分

    “#”:代表0-多个部分,路由键为#,则代表接受所有信息

  2. headers 性能太差 不做学习



版权声明:本文为qq_67548292原创文章,遵循 CC 4.0 BY-SA 版权协议,转载请附上原文出处链接和本声明。