RabbitMQ学习札记 – 记忆角落

RabbitMQ学习札记

/ 0评 / 0

RabbitMQ学习札记

1.MQ概述

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

2.MQ 的优势和劣势

优势:

  1. 应用解耦:提高系统容错性和可维护性
  2. 异步提速:提升用户体验和系统吞吐量
  3. 削峰填谷:提高系统稳定性

如订单系统,在下单的时候就会往数据库写数据。但是数据库只能支撑每秒1000左右的并发写入,并发量再高就容易宕机。低峰期的时候并发也就100多个,但是在高峰期时候,并发量会突然激增到5000以上,这个时候数据库肯定卡死了。

消息被MQ保存起来了,然后系统就可以按照自己的消费能力来消费,比如每秒1000个数据,这样慢慢写入数据库,这样就不会卡死数据库了。

但是使用了MQ之后,限制消费消息的速度为1000,但是这样一来,高峰期产生的数据势必会被积压在MQ中,高峰就被“削”掉了。但是因为消息积压,在高峰期过后的一段时间内,消费消息的速度还是会维持在1000QPS,直到消费完积压的消息,这就叫做“填谷”

03

劣势:

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

3.AMQP 和 JMS

MQ是消息通信的模型;实现MQ的大致有两种主流方式:AMQP、JMS。

3.1. AMQP

AMQP是一种协议,更准确的说是一种binary wire-level protocol(链接协议)。这是其和JMS的本质差别,AMQP不从API层进行限定,而是直接定义网络交换的数据格式。

3.2. JMS

JMS即Java消息服务(JavaMessage Service)应用程序接口,是一个Java平台中关于面向消息中间件(MOM)的API,用于在两个应用程序之间,或分布式系统中发送消息,进行异步通信。

3.3. AMQP 与 JMS 区别

4.常见的 MQ 产品

目前业界有很多的 MQ 产品,例如 RabbitMQ、RocketMQ、ActiveMQ、Kafka、ZeroMQ、MetaMq等,也有直接使用 Redis 充当消息队列的案例。

市场上常见的消息队列有如下:

image-20210616235052803

5.RabbitMQ 简介

RabbitMQ是由erlang语言开发,基于AMQP(Advanced Message Queue 高级消息队列协议)协议实现的消息队列,它是一种应用程序之间的通信方法,消息队列在分布式系统开发中应用非常广泛。

RabbitMQ官方地址:http://www.rabbitmq.com/

RabbitMQ提供了6种模式:简单模式,work模式,Publish/Subscribe发布与订阅模式,Routing路由模式,Topics主题模式,RPC远程调用模式(远程调用,不太算MQ;暂不作介绍);

官网对应模式介绍:https://www.rabbitmq.com/getstarted.html

1555988678324

6.安装及配置RabbitMQ

6.1 安装依赖环境

安装依赖环境:

yum install build-essential openssl openssl-devel unixODBC unixODBC-devel make gcc gcc-c++ kernel-devel m4 ncurses-devel tk tc xz

根据官方建议要先安装:

yum install socat

yum install logrotate

再安装erlang

下载:https://packages.erlang-solutions.com/erlang/rpm/centos/7/x86_64/esl-erlang_22.3.2-1~centos~7_amd64.rpm

安装:rpm -Uvh esl-erlang_22.3.2-1_centos_7_amd64.rpm --force --nodeps

最后安装rabbitmq

这个页面是rabbitmq对应的erlang版本查询网站:https://www.rabbitmq.com/which-erlang.html

下载:https://github.com/rabbitmq/rabbitmq-server/releases/download/v3.8.3/rabbitmq-server-3.8.3-1.el7.noarch.rpm

安装:rpm -ivh rabbitmq-server-3.8.3-1.el7.noarch.rpm

安装路径默认在:

/usr/lib/rabbitmq/lib/rabbitmq_server-3.8.3/ebin

6.2开启管理界面及配置

# 开启rabbitmq management及远程登录
rabbitmq-plugins enable rabbitmq_management

# 设置guest远程访问:
vi /usr/lib/rabbitmq/lib/rabbitmq_server-3.8.3/ebin/rabbit.app
# loopback_users里的<<"guest">>删除,重启rabbitmq服务
# 比如修改密码、配置等等,例如:loopback_users 中的 <<"guest">>,只保留guest

要访问RabbitMQ的管理面板,请使用您最喜爱的Web浏览器并打开以下URL:

http://Your_Server_IP:15672/

6.3 启动停止服务

service rabbitmq-server start # 启动服务
service rabbitmq-server stop # 停止服务
service rabbitmq-server restart # 重启服务

6.4 设置配置文件

如果不知道配置文件在哪,可以查找:

[root@wydsj-gpapp04 ~]# find / -name rabbitmq-defaults
/usr/lib/rabbitmq/bin/rabbitmq-defaults
/usr/lib/rabbitmq/lib/rabbitmq_server-3.8.3/sbin/rabbitmq-defaults

6.5 配置虚拟主机及用户

角色说明

1、 超级管理员(administrator)

可登陆管理控制台,可查看所有的信息,并且可以对用户,策略(policy)进行操作。

2、 监控者(monitoring)

可登陆管理控制台,同时可以查看rabbitmq节点的相关信息(进程数,内存使用情况,磁盘使用情况等)

3、 策略制定者(policymaker)

可登陆管理控制台, 同时可以对policy进行管理。但无法查看节点的相关信息(上图红框标识的部分)。

4、 普通管理者(management)

仅可登陆管理控制台,无法看到节点信息,也无法对策略进行管理。

5、 其他

无法登陆管理控制台,通常就是普通的生产者和消费者。

配置有两种方式:
  1. 命令行
  2. 界面

6.5.1 命令行方式:

创建用户:

创建用户名hcxtadmin,密码hcxtadmin123 的用户

rabbitmqctl add_user hcxtadmin hcxtadmin123
设置超级管理员

设置hcxtadmin为超级管理员

rabbitmqctl set_user_tags hcxtadmin administrator
授权远程访问(也可以登录后,可视化配置)
rabbitmqctl set_permissions -p / hcxtadmin "." "." ".*"

image-20210617011801106

查看用户列表:
#查看用户列表
rabbitmqctl  list_users
修改对应用户密码
rabbitmqctl  change_password  username  'newpasswd'

6.5.2 界面方式:

创建用户并且超级管理员:

创建用户名hcxtuser,密码hcxtuser123!@#的用户

image-20210617012715084

授权远程访问:

点击用户名称进行授权

image-20210617013004618

6.5.3.Virtual Hosts配置

像mysql拥有数据库的概念并且可以指定用户对库和表等操作的权限。RabbitMQ也有类似的权限管理;在RabbitMQ中可以虚拟消息服务器Virtual Host,每个Virtual Hosts相当于一个相对独立的RabbitMQ服务器,每个VirtualHost之间是相互隔离的。exchange、queue、message不能互通。 相当于mysql的db。Virtual Name一般以/开头。

1-创建Virtual Hosts

如下图:

image-20210617013347343

2-设置Virtual Hosts权限

image-20210617013607062

image-20210617013804832

6.6.常用配置说明:

tcp_listerners    #设置rabbimq的监听端口,默认为[5672]。
disk_free_limit     #磁盘低水位线,若磁盘容量低于指定值则停止接收数据,默认值为{mem_relative, 1.0},即与内存相关联1:1,也可定制为多少byte.
vm_memory_high_watermark    #设置内存低水位线,若低于该水位线,则开启流控机制,默认值是0.4,即内存总量的40%。
hipe_compile     #将部分rabbimq代码用High Performance Erlang compiler编译,可提升性能,该参数是实验性,若出现erlang vm segfaults,应关掉。
force_fine_statistics    #该参数属于rabbimq_management,若为true则进行精细化的统计,但会影响性能。
frame_max     #包大小,若包小则低延迟,若包则高吞吐,默认是131072=128K。
heartbeat     #客户端与服务端心跳间隔,设置为0则关闭心跳,默认是60秒。

7.RabbitMQ入门

7.1. 搭建示例工程

7.1.1. 创建工程

image-20210620201310189

7.1.2. 添加依赖

<dependencies>
    <dependency>
        <groupId>com.rabbitmq</groupId>
        <artifactId>amqp-client</artifactId>
        <version>5.6.0</version>
    </dependency>
</dependencies>


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

7.2. 编写生产者

package com.glj.rabitmq.simple;

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

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: GuoLiangJun
 * @Date: 2021/6/20
 * @Time: 20:16
 */
public class Producer {

    static final String QUEUE_NAME = "simple_queue";

    public static void main(String[] args) {
        // 创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        // 主机地址;默认为 localhost、127.0.0.1
        connectionFactory.setHost("127.0.0.1");
        // 连接端口;默认为 5672
        connectionFactory.setPort(5672);
        // 虚拟主机名称;默认为 /
        connectionFactory.setVirtualHost("/hcxtuser");
        // 连接用户名;默认为guest
        connectionFactory.setUsername("hcxtuser");
        // 连接密码;默认为guest
        connectionFactory.setPassword("hcxtuser123!@#");

        try(
                // 创建连接
                Connection connection = connectionFactory.newConnection();
                // 创建频道
                Channel channel = connection.createChannel()
        ){
            // 声明(创建)队列
            /**
             * 参数1:队列名称
             * 参数2:是否定义持久化队列
             * 参数3:是否独占本次连接
             * 参数4:是否在不使用的时候自动删除队列
             * 参数5:队列其它参数
             */
            channel.queueDeclare(QUEUE_NAME,true,false,false,null);
            // 要发送的信息
            String mes = "hello rabitmq";
            /**
             * 参数1:交换机名称,如果没有指定则使用默认Default Exchage
             * 参数2:路由key,简单模式可以传递队列名称
             * 参数3:消息其它属性
             * 参数4:消息内容
             */
            channel.basicPublish("", QUEUE_NAME, null, mes.getBytes());
            System.out.println("已发送消息:" + mes);


        }catch (Exception e){
            e.printStackTrace();
        }
    }

}

在执行上述的消息发送之后;可以登录rabbitMQ的管理控制台,可以发现队列和其消息:

image-20210620203458596

7.3. 编写消费者

package com.glj.rabitmq.simple;

import com.rabbitmq.client.*;

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

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: GuoLiangJun
 * @Date: 2021/6/20
 * @Time: 20:44
 */
public class Consumer {

    static final String QUEUE_NAME = "simple_queue";

    public static void main(String[] args) throws Exception {

        // 创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        // 主机地址;默认为 localhost、127.0.0.1
        connectionFactory.setHost("127.0.0.1");
        // 连接端口;默认为 5672
        connectionFactory.setPort(5672);
        // 虚拟主机名称;默认为 /
        connectionFactory.setVirtualHost("/hcxtuser");
        // 连接用户名;默认为guest
        connectionFactory.setUsername("hcxtuser");
        // 连接密码;默认为guest
        connectionFactory.setPassword("hcxtuser123!@#");

        // 创建连接
        Connection connection = connectionFactory.newConnection
                ();
        // 创建频道
        Channel channel = connection.createChannel();
        // 声明(创建)队列
        /**
         * 参数1:队列名称
         * 参数2:是否定义持久化队列
         * 参数3:是否独占本次连接
         * 参数4:是否在不使用的时候自动删除队列
         * 参数5:队列其它参数
         */
        channel.queueDeclare(QUEUE_NAME,true,false,false,null);

        DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
            @Override
            /**
             回调方法,当介绍消息后,会自动执行该方法
             * consumerTag 消息者标签,在channel.basicConsume时候可以指定
             * envelope 消息包的内容,可从中获取消息id,消息routingkey,交换机,消息和重传标志(收到消息失败后是否需要重新发送)
             * properties 属性信息
             * body 消息
             */
            public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                //消息者标签
                System.out.println("consumerTag:"+consumerTag);
                //路由key
                System.out.println("路由key为:" + envelope.getRoutingKey());
                //交换机
                System.out.println("交换机为:" + envelope.getExchange());
                //消息id
                System.out.println("消息id为:" + envelope.getDeliveryTag());
                //收到的消息
                System.out.println("接收到的消息为:" + new String(body, "utf-8"));

            }
        };
        //监听消息
        /**
         * 参数1:队列名称
         * 参数2:是否自动确认,设置为true为表示消息接收到自动向mq回复接收到了,mq接收到回复会删除消息,设置为false则需要手动确认
         * 参数3:消息接收到后回调
         */
        channel.basicConsume(QUEUE_NAME,true,defaultConsumer);

        //不关闭资源,应该一直监听消息
        //channel.close();
        //connection.close();
    }

}

7.4. 小结

上述的入门案例中中其实使用的是如下的简单模式:

1555991074575

在上图的模型中,有以下概念:

8.AMQP

8.1. 相关概念介绍

AMQP 一个提供统一消息服务的应用层标准高级消息队列协议,是应用层协议的一个开放标准,为面向消息的中间件设计。

AMQP是一个二进制协议,拥有一些现代化特点:多信道、协商式,异步,安全,扩平台,中立,高效。

RabbitMQ是AMQP协议的Erlang的实现。

概念 说明
连接Connection 一个网络连接,比如TCP/IP套接字连接。
会话Session 端点之间的命名对话。在一个会话上下文中,保证“恰好传递一次”。
信道Channel 多路复用连接中的一条独立的双向数据流通道。为会话提供物理传输介质。
客户端Client AMQP连接或者会话的发起者。AMQP是非对称的,客户端生产和消费消息,服务器存储和路由这些消息。
服务节点Broker 消息中间件的服务节点;一般情况下可以将一个RabbitMQ Broker看作一台RabbitMQ 服务器。
端点 AMQP对话的任意一方。一个AMQP连接包括两个端点(一个是客户端,一个是服务器)。
消费者Consumer 一个从消息队列里请求消息的客户端程序。
生产者Producer 一个向交换机发布消息的客户端应用程序。

8.1.RabbitMQ运转流程

在入门案例中:

9.RabbitMQ工作模式

9.1. Work queues工作队列模式

9.1.1. 模式说明

1556009144848

Work Queues与入门程序的简单模式相比,多了一个或一些消费端,多个消费端共同消费同一个队列中的消息。

应用场景:对于 任务过重或任务较多情况使用工作队列可以提高任务处理的速度。

9.1.2.代码

Work Queues与入门程序的简单模式的代码是几乎一样的;可以完全复制,并复制多一个消费者进行多个消费者同时消费消息的测试。

创建connection的工具类ConnectionUtil
package com.glj.rabitmq.util;

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

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: GuoLiangJun
 * @Date: 2021/6/20
 * @Time: 21:09
 */
public class ConnectionUtil {

    public static Connection getConnection() throws Exception {
        // 创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        // 主机地址;默认为 localhost、127.0.0.1
        connectionFactory.setHost("127.0.0.1");
        // 连接端口;默认为 5672
        connectionFactory.setPort(5672);
        // 虚拟主机名称;默认为 /
        connectionFactory.setVirtualHost("/hcxtuser");
        // 连接用户名;默认为guest
        connectionFactory.setUsername("hcxtuser");
        // 连接密码;默认为guest
        connectionFactory.setPassword("hcxtuser123!@#");

        //创建连接
        return connectionFactory.newConnection();
    }

}

生产者:
package com.glj.rabitmq.work;

import com.glj.rabitmq.util.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: GuoLiangJun
 * @Date: 2021/6/20
 * @Time: 20:16
 */
public class Producer {

    static final String QUEUE_NAME = "work_queue";

    public static void main(String[] args) {
        try(
                // 创建连接
                Connection connection = ConnectionUtil.getConnection();
                // 创建频道
                Channel channel = connection.createChannel()
        ){
            // 声明(创建)队列
            /**
             * 参数1:队列名称
             * 参数2:是否定义持久化队列
             * 参数3:是否独占本次连接
             * 参数4:是否在不使用的时候自动删除队列
             * 参数5:队列其它参数
             */
            channel.queueDeclare(QUEUE_NAME,true,false,false,null);

            for (int i = 1; i < 20; i++) {
                // 要发送的信息
                String mes = "rabitmq-work模式--"+i;
                /**
                 * 参数1:交换机名称,如果没有指定则使用默认Default Exchage
                 * 参数2:路由key,简单模式可以传递队列名称
                 * 参数3:消息其它属性
                 * 参数4:消息内容
                 */
                channel.basicPublish("", QUEUE_NAME, null, mes.getBytes());
                System.out.println("已发送消息:" + mes);
            }
        }catch (Exception e){
            e.printStackTrace();
        }
    }

}

消费者1:
package com.glj.rabitmq.work;

import com.glj.rabitmq.util.ConnectionUtil;
import com.rabbitmq.client.*;

import java.io.IOException;

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: GuoLiangJun
 * @Date: 2021/6/20
 * @Time: 20:44
 */
public class Consumer1 {

    static final String QUEUE_NAME = "work_queue";

    public static void main(String[] args) throws Exception {

        // 创建连接
        Connection connection = ConnectionUtil.getConnection();
        // 创建频道
        Channel channel = connection.createChannel();
        // 声明(创建)队列
        /**
         * 参数1:队列名称
         * 参数2:是否定义持久化队列
         * 参数3:是否独占本次连接
         * 参数4:是否在不使用的时候自动删除队列
         * 参数5:队列其它参数
         */
        channel.queueDeclare(QUEUE_NAME,true,false,false,null);

        DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
            @Override
            /**
             回调方法,当介绍消息后,会自动执行该方法
             * consumerTag 消息者标签,在channel.basicConsume时候可以指定
             * envelope 消息包的内容,可从中获取消息id,消息routingkey,交换机,消息和重传标志(收到消息失败后是否需要重新发送)
             * properties 属性信息
             * body 消息
             */
            public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                //收到的消息
                System.out.println("接收到的消息为:" + new String(body, "utf-8"));

            }
        };
        //监听消息
        /**
         * 参数1:队列名称
         * 参数2:是否自动确认,设置为true为表示消息接收到自动向mq回复接收到了,mq接收到回复会删除消息,设置为false则需要手动确认
         * 参数3:消息接收到后回调
         */
        channel.basicConsume(QUEUE_NAME,true,defaultConsumer);

        //不关闭资源,应该一直监听消息
        //channel.close();
        //connection.close();
    }

}

消费者2:
package com.glj.rabitmq.work;

import com.glj.rabitmq.util.ConnectionUtil;
import com.rabbitmq.client.*;

import java.io.IOException;

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: GuoLiangJun
 * @Date: 2021/6/20
 * @Time: 20:44
 */
public class Consumer2 {

    static final String QUEUE_NAME = "work_queue";

    public static void main(String[] args) throws Exception {
        // 创建连接
        Connection connection = ConnectionUtil.getConnection();
        // 创建频道
        Channel channel = connection.createChannel();
        // 声明(创建)队列
        /**
         * 参数1:队列名称
         * 参数2:是否定义持久化队列
         * 参数3:是否独占本次连接
         * 参数4:是否在不使用的时候自动删除队列
         * 参数5:队列其它参数
         */
        channel.queueDeclare(QUEUE_NAME,true,false,false,null);

        DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
            @Override
            /**
             回调方法,当介绍消息后,会自动执行该方法
             * consumerTag 消息者标签,在channel.basicConsume时候可以指定
             * envelope 消息包的内容,可从中获取消息id,消息routingkey,交换机,消息和重传标志(收到消息失败后是否需要重新发送)
             * properties 属性信息
             * body 消息
             */
            public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                //收到的消息
                System.out.println("接收到的消息为:" + new String(body, "utf-8"));

            }
        };
        //监听消息
        /**
         * 参数1:队列名称
         * 参数2:是否自动确认,设置为true为表示消息接收到自动向mq回复接收到了,mq接收到回复会删除消息,设置为false则需要手动确认
         * 参数3:消息接收到后回调
         */
        channel.basicConsume(QUEUE_NAME,true,defaultConsumer);

        //不关闭资源,应该一直监听消息
        //channel.close();
        //connection.close();
    }

}

小结

在一个队列中如果有多个消费者,那么消费者之间对于同一个消息的关系是竞争的关系。

9.2.订阅模式类型说明

前面2个案例中,只有3个角色:

订阅模式示例图:

1556014499573

而在订阅模型中,多了一个exchange角色,而且过程略有变化:

Exchange(交换机)只负责转发消息,不具备存储消息的能力,因此如果没有任何队列与Exchange绑定,或者没有符合路由规则的队列,那么消息会丢失!

9.3.Publish/Subscribe发布与订阅模式(fanout)

模式说明

1556010329032

发布订阅模式:
1、每个消费者监听自己的队列。
2、生产者将消息发给broker,由交换机将消息转发到绑定此交换机的每个队列,每个绑定交换机的队列都将接收
到消息

代码:

1.生产者:
package com.glj.rabitmq.ps;

import com.glj.rabitmq.util.ConnectionUtil;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: GuoLiangJun
 * @Date: 2021/6/20
 * @Time: 20:16
 */
public class Producer {

    //交换机名称
    static final String FANOUT_EXCHAGE = "fanout_exchange";
    //队列名称
    static final String FANOUT_QUEUE_1 = "fanout_queue_1";
    //队列名称
    static final String FANOUT_QUEUE_2 = "fanout_queue_2";

    public static void main(String[] args) {
        try(
                // 创建连接
                Connection connection = ConnectionUtil.getConnection();
                // 创建频道
                Channel channel = connection.createChannel()
        ){
            /**
                声明交换机
                exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments)
               参数:
                1. exchange:交换机名称
                2. type:交换机类型
                    DIRECT("direct"),:定向
                    FANOUT("fanout"),:扇形(广播),发送消息到每一个与之绑定队列。
                    TOPIC("topic"),通配符的方式
                    HEADERS("headers");参数匹配

                3. durable:是否持久化
                4. autoDelete:自动删除
                5. internal:内部使用。 一般false
                6. arguments:参数
             */
            channel.exchangeDeclare(FANOUT_EXCHAGE, BuiltinExchangeType.FANOUT,true,false,false,null);

            // 声明(创建)队列
            /**
             * 参数1:队列名称
             * 参数2:是否定义持久化队列
             * 参数3:是否独占本次连接
             * 参数4:是否在不使用的时候自动删除队列
             * 参数5:队列其它参数
             */
            channel.queueDeclare(FANOUT_QUEUE_1,true,false,false,null);
            channel.queueDeclare(FANOUT_QUEUE_2,true,false,false,null);

            /**
             * 队列绑定交换机
            queueBind(String queue, String exchange, String routingKey)
            参数:
                1. queue:队列名称
                2. exchange:交换机名称
                3. routingKey:路由键,绑定规则
                    如果交换机的类型为fanout ,routingKey设置为""
             */
            channel.queueBind(FANOUT_QUEUE_1,FANOUT_EXCHAGE,"");
            channel.queueBind(FANOUT_QUEUE_2,FANOUT_EXCHAGE,"");

            for (int i = 1; i < 20; i++) {
                // 要发送的信息
                String mes = "rabitmq-work模式--"+i;
                /**
                 * 参数1:交换机名称,如果没有指定则使用默认Default Exchage
                 * 参数2:路由key,简单模式可以传递队列名称
                 * 参数3:消息其它属性
                 * 参数4:消息内容
                 */
                channel.basicPublish(FANOUT_EXCHAGE, "", null, mes.getBytes());
                System.out.println("已发送消息:" + mes);
            }
        }catch (Exception e){
            e.printStackTrace();
        }
    }

}

2.消费者1:
package com.glj.rabitmq.ps;

import com.glj.rabitmq.util.ConnectionUtil;
import com.rabbitmq.client.*;

import java.io.IOException;

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: GuoLiangJun
 * @Date: 2021/6/20
 * @Time: 20:44
 */
public class Consumer1 {

    //交换机名称
    static final String FANOUT_EXCHAGE = "fanout_exchange";
    //队列名称
    static final String FANOUT_QUEUE_1 = "fanout_queue_1";

    public static void main(String[] args) throws Exception {

        // 创建连接
        Connection connection = ConnectionUtil.getConnection();
        // 创建频道
        Channel channel = connection.createChannel();

        /**
         声明交换机
         exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments)
         参数:
         1. exchange:交换机名称
         2. type:交换机类型
         DIRECT("direct"),:定向
         FANOUT("fanout"),:扇形(广播),发送消息到每一个与之绑定队列。
         TOPIC("topic"),通配符的方式
         HEADERS("headers");参数匹配

         3. durable:是否持久化
         4. autoDelete:自动删除
         5. internal:内部使用。 一般false
         6. arguments:参数
         */
        channel.exchangeDeclare(FANOUT_EXCHAGE, BuiltinExchangeType.FANOUT,true,false,false,null);

        // 声明(创建)队列
        /**
         * 参数1:队列名称
         * 参数2:是否定义持久化队列
         * 参数3:是否独占本次连接
         * 参数4:是否在不使用的时候自动删除队列
         * 参数5:队列其它参数
         */
        channel.queueDeclare(FANOUT_QUEUE_1,true,false,false,null);

        /**
         * 队列绑定交换机
         queueBind(String queue, String exchange, String routingKey)
         参数:
         1. queue:队列名称
         2. exchange:交换机名称
         3. routingKey:路由键,绑定规则
         如果交换机的类型为fanout ,routingKey设置为""
         */
        channel.queueBind(FANOUT_QUEUE_1,FANOUT_EXCHAGE,"");

        DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
            @Override
            /**
             回调方法,当介绍消息后,会自动执行该方法
             * consumerTag 消息者标签,在channel.basicConsume时候可以指定
             * envelope 消息包的内容,可从中获取消息id,消息routingkey,交换机,消息和重传标志(收到消息失败后是否需要重新发送)
             * properties 属性信息
             * body 消息
             */
            public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                //收到的消息
                System.out.println("消费者1-接收到的消息为:" + new String(body, "utf-8"));

            }
        };
        //监听消息
        /**
         * 参数1:队列名称
         * 参数2:是否自动确认,设置为true为表示消息接收到自动向mq回复接收到了,mq接收到回复会删除消息,设置为false则需要手动确认
         * 参数3:消息接收到后回调
         */
        channel.basicConsume(FANOUT_QUEUE_1,true,defaultConsumer);

        //不关闭资源,应该一直监听消息
        //channel.close();
        //connection.close();
    }

}

3.消费者2:
package com.glj.rabitmq.ps;

import com.glj.rabitmq.util.ConnectionUtil;
import com.rabbitmq.client.*;

import java.io.IOException;

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: GuoLiangJun
 * @Date: 2021/6/20
 * @Time: 20:44
 */
public class Consumer2 {

    //交换机名称
    static final String FANOUT_EXCHAGE = "fanout_exchange";
    //队列名称
    static final String FANOUT_QUEUE_1 = "fanout_queue_2";

    public static void main(String[] args) throws Exception {

        // 创建连接
        Connection connection = ConnectionUtil.getConnection();
        // 创建频道
        Channel channel = connection.createChannel();

        /**
         声明交换机
         exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments)
         参数:
         1. exchange:交换机名称
         2. type:交换机类型
         DIRECT("direct"),:定向
         FANOUT("fanout"),:扇形(广播),发送消息到每一个与之绑定队列。
         TOPIC("topic"),通配符的方式
         HEADERS("headers");参数匹配

         3. durable:是否持久化
         4. autoDelete:自动删除
         5. internal:内部使用。 一般false
         6. arguments:参数
         */
        channel.exchangeDeclare(FANOUT_EXCHAGE, BuiltinExchangeType.FANOUT,true,false,false,null);

        // 声明(创建)队列
        /**
         * 参数1:队列名称
         * 参数2:是否定义持久化队列
         * 参数3:是否独占本次连接
         * 参数4:是否在不使用的时候自动删除队列
         * 参数5:队列其它参数
         */
        channel.queueDeclare(FANOUT_QUEUE_1,true,false,false,null);

        /**
         * 队列绑定交换机
         queueBind(String queue, String exchange, String routingKey)
         参数:
         1. queue:队列名称
         2. exchange:交换机名称
         3. routingKey:路由键,绑定规则
         如果交换机的类型为fanout ,routingKey设置为""
         */
        channel.queueBind(FANOUT_QUEUE_1,FANOUT_EXCHAGE,"");

        DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
            @Override
            /**
             回调方法,当介绍消息后,会自动执行该方法
             * consumerTag 消息者标签,在channel.basicConsume时候可以指定
             * envelope 消息包的内容,可从中获取消息id,消息routingkey,交换机,消息和重传标志(收到消息失败后是否需要重新发送)
             * properties 属性信息
             * body 消息
             */
            public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                //收到的消息
                System.out.println("消费者2-接收到的消息为:" + new String(body, "utf-8"));

            }
        };
        //监听消息
        /**
         * 参数1:队列名称
         * 参数2:是否自动确认,设置为true为表示消息接收到自动向mq回复接收到了,mq接收到回复会删除消息,设置为false则需要手动确认
         * 参数3:消息接收到后回调
         */
        channel.basicConsume(FANOUT_QUEUE_1,true,defaultConsumer);

        //不关闭资源,应该一直监听消息
        //channel.close();
        //connection.close();
    }

}

4.小结

交换机需要与队列进行绑定,绑定之后;一个消息可以被多个消费者都收到。

5.发布订阅模式与工作队列模式的区别

1、工作队列模式不用定义交换机,而发布/订阅模式需要定义交换机。

2、发布/订阅模式的生产方是面向交换机发送消息,工作队列模式的生产方是面向队列发送消息(底层使用默认交换机)。

3、发布/订阅模式需要设置队列和交换机的绑定,工作队列模式不需要设置,实际上工作队列模式会将队列绑 定到默认的交换机 .

9.4. Routing路由模式(direct)

模式说明

路由模式特点:

1556029284397

图解:

代码

在编码上与 Publish/Subscribe发布与订阅模式 的区别是交换机的类型为:Direct,还有队列绑定交换机的时候需要指定routing key。

1.生产者:
package com.glj.rabitmq.routing;

import com.glj.rabitmq.util.ConnectionUtil;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: GuoLiangJun
 * @Date: 2021/6/20
 * @Time: 20:16
 */
public class Producer {

    //交换机名称
    static final String FANOUT_EXCHAGE = "direct_exchange";
    //队列名称
    static final String FANOUT_QUEUE_1 = "direct_queue_1";
    //队列名称
    static final String FANOUT_QUEUE_2 = "direct_queue_2";

    public static void main(String[] args) {
        try(
                // 创建连接
                Connection connection = ConnectionUtil.getConnection();
                // 创建频道
                Channel channel = connection.createChannel()
        ){
            /**
                声明交换机
                exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments)
               参数:
                1. exchange:交换机名称
                2. type:交换机类型
                    DIRECT("direct"),:定向
                    FANOUT("fanout"),:扇形(广播),发送消息到每一个与之绑定队列。
                    TOPIC("topic"),通配符的方式
                    HEADERS("headers");参数匹配

                3. durable:是否持久化
                4. autoDelete:自动删除
                5. internal:内部使用。 一般false
                6. arguments:参数
             */
            channel.exchangeDeclare(FANOUT_EXCHAGE, BuiltinExchangeType.DIRECT,true,false,false,null);

            // 声明(创建)队列
            /**
             * 参数1:队列名称
             * 参数2:是否定义持久化队列
             * 参数3:是否独占本次连接
             * 参数4:是否在不使用的时候自动删除队列
             * 参数5:队列其它参数
             */
            channel.queueDeclare(FANOUT_QUEUE_1,true,false,false,null);
            channel.queueDeclare(FANOUT_QUEUE_2,true,false,false,null);

            /**
             * 队列绑定交换机
            queueBind(String queue, String exchange, String routingKey)
            参数:
                1. queue:队列名称
                2. exchange:交换机名称
                3. routingKey:路由键,绑定规则
                    如果交换机的类型为fanout ,routingKey设置为""
             */
            channel.queueBind(FANOUT_QUEUE_1,FANOUT_EXCHAGE,"error");
            channel.queueBind(FANOUT_QUEUE_2,FANOUT_EXCHAGE,"error");
            channel.queueBind(FANOUT_QUEUE_2,FANOUT_EXCHAGE,"info");
            channel.queueBind(FANOUT_QUEUE_2,FANOUT_EXCHAGE,"warn");

            // 要发送的信息
            String mes = "rabitmq---info";
            /**
             * 参数1:交换机名称,如果没有指定则使用默认Default Exchage
             * 参数2:路由key,简单模式可以传递队列名称
             * 参数3:消息其它属性
             * 参数4:消息内容
             */
            channel.basicPublish(FANOUT_EXCHAGE, "info", null, mes.getBytes());
            System.out.println("已发送消息:" + mes);

            // 要发送的信息
            mes = "rabitmq---error";
            channel.basicPublish(FANOUT_EXCHAGE, "error", null, mes.getBytes());
            System.out.println("已发送消息:" + mes);
        }catch (Exception e){
            e.printStackTrace();
        }
    }

}

2.消费者1:
package com.glj.rabitmq.routing;

import com.glj.rabitmq.util.ConnectionUtil;
import com.rabbitmq.client.*;

import java.io.IOException;

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: GuoLiangJun
 * @Date: 2021/6/20
 * @Time: 20:44
 */
public class Consumer1 {

    //交换机名称
    static final String FANOUT_EXCHAGE = "direct_exchange";
    //队列名称
    static final String FANOUT_QUEUE_1 = "direct_queue_1";

    public static void main(String[] args) throws Exception {

        // 创建连接
        Connection connection = ConnectionUtil.getConnection();
        // 创建频道
        Channel channel = connection.createChannel();

        /**
         声明交换机
         exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments)
         参数:
         1. exchange:交换机名称
         2. type:交换机类型
         DIRECT("direct"),:定向
         FANOUT("fanout"),:扇形(广播),发送消息到每一个与之绑定队列。
         TOPIC("topic"),通配符的方式
         HEADERS("headers");参数匹配

         3. durable:是否持久化
         4. autoDelete:自动删除
         5. internal:内部使用。 一般false
         6. arguments:参数
         */
        channel.exchangeDeclare(FANOUT_EXCHAGE, BuiltinExchangeType.DIRECT,true,false,false,null);

        // 声明(创建)队列
        /**
         * 参数1:队列名称
         * 参数2:是否定义持久化队列
         * 参数3:是否独占本次连接
         * 参数4:是否在不使用的时候自动删除队列
         * 参数5:队列其它参数
         */
        channel.queueDeclare(FANOUT_QUEUE_1,true,false,false,null);

        /**
         * 队列绑定交换机
         queueBind(String queue, String exchange, String routingKey)
         参数:
         1. queue:队列名称
         2. exchange:交换机名称
         3. routingKey:路由键,绑定规则
         如果交换机的类型为fanout ,routingKey设置为""
         */
        channel.queueBind(FANOUT_QUEUE_1,FANOUT_EXCHAGE,"error");

        DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
            @Override
            /**
             回调方法,当介绍消息后,会自动执行该方法
             * consumerTag 消息者标签,在channel.basicConsume时候可以指定
             * envelope 消息包的内容,可从中获取消息id,消息routingkey,交换机,消息和重传标志(收到消息失败后是否需要重新发送)
             * properties 属性信息
             * body 消息
             */
            public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                //收到的消息
                System.out.println("消费者1-接收到的消息为:" + new String(body, "utf-8"));

            }
        };
        //监听消息
        /**
         * 参数1:队列名称
         * 参数2:是否自动确认,设置为true为表示消息接收到自动向mq回复接收到了,mq接收到回复会删除消息,设置为false则需要手动确认
         * 参数3:消息接收到后回调
         */
        channel.basicConsume(FANOUT_QUEUE_1,true,defaultConsumer);

        //不关闭资源,应该一直监听消息
        //channel.close();
        //connection.close();
    }

}

3.消费者2:
package com.glj.rabitmq.routing;

import com.glj.rabitmq.util.ConnectionUtil;
import com.rabbitmq.client.*;

import java.io.IOException;

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: GuoLiangJun
 * @Date: 2021/6/20
 * @Time: 20:44
 */
public class Consumer2 {

    //交换机名称
    static final String FANOUT_EXCHAGE = "direct_exchange";
    //队列名称
    static final String FANOUT_QUEUE_1 = "direct_queue_2";

    public static void main(String[] args) throws Exception {

        // 创建连接
        Connection connection = ConnectionUtil.getConnection();
        // 创建频道
        Channel channel = connection.createChannel();

        /**
         声明交换机
         exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments)
         参数:
         1. exchange:交换机名称
         2. type:交换机类型
         DIRECT("direct"),:定向
         FANOUT("fanout"),:扇形(广播),发送消息到每一个与之绑定队列。
         TOPIC("topic"),通配符的方式
         HEADERS("headers");参数匹配

         3. durable:是否持久化
         4. autoDelete:自动删除
         5. internal:内部使用。 一般false
         6. arguments:参数
         */
        channel.exchangeDeclare(FANOUT_EXCHAGE, BuiltinExchangeType.DIRECT,true,false,false,null);

        // 声明(创建)队列
        /**
         * 参数1:队列名称
         * 参数2:是否定义持久化队列
         * 参数3:是否独占本次连接
         * 参数4:是否在不使用的时候自动删除队列
         * 参数5:队列其它参数
         */
        channel.queueDeclare(FANOUT_QUEUE_1,true,false,false,null);

        /**
         * 队列绑定交换机
         queueBind(String queue, String exchange, String routingKey)
         参数:
         1. queue:队列名称
         2. exchange:交换机名称
         3. routingKey:路由键,绑定规则
         如果交换机的类型为fanout ,routingKey设置为""
         */
        channel.queueBind(FANOUT_QUEUE_1,FANOUT_EXCHAGE,"error");
        channel.queueBind(FANOUT_QUEUE_1,FANOUT_EXCHAGE,"info");
        channel.queueBind(FANOUT_QUEUE_1,FANOUT_EXCHAGE,"warn");

        DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
            @Override
            /**
             回调方法,当介绍消息后,会自动执行该方法
             * consumerTag 消息者标签,在channel.basicConsume时候可以指定
             * envelope 消息包的内容,可从中获取消息id,消息routingkey,交换机,消息和重传标志(收到消息失败后是否需要重新发送)
             * properties 属性信息
             * body 消息
             */
            public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                //收到的消息
                System.out.println("消费者2-接收到的消息为:" + new String(body, "utf-8"));

            }
        };
        //监听消息
        /**
         * 参数1:队列名称
         * 参数2:是否自动确认,设置为true为表示消息接收到自动向mq回复接收到了,mq接收到回复会删除消息,设置为false则需要手动确认
         * 参数3:消息接收到后回调
         */
        channel.basicConsume(FANOUT_QUEUE_1,true,defaultConsumer);

        //不关闭资源,应该一直监听消息
        //channel.close();
        //connection.close();
    }

}

小结

Routing模式要求队列在绑定交换机时要指定routing key,消息会转发到符合routing key的队列。

9.5.Topics通配符模式

模式说明

Topic类型与Direct相比,都是可以根据RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定Routing key 的时候使用通配符

Routingkey 一般都是有一个或多个单词组成,多个单词之间以”.”分割,例如: item.insert

通配符规则:

#:匹配一个或多个词

*:匹配不多不少恰好1个词

举例:

item.#:能够匹配item.insert.abc 或者 item.insert

item.*:只能匹配item.insert

1556031362048

1556031519931

图解:

代码

在编码上与 路由模式 的区别是交换机的类型为:TOPIC,还有消费者不需要绑定声明队列、不需要队列绑定交换机。

1.生产者

使用topic类型的Exchange,发送消息的routing key有3种: #.erroruser.**.*

package com.glj.rabitmq.topic;

import com.glj.rabitmq.util.ConnectionUtil;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: GuoLiangJun
 * @Date: 2021/6/20
 * @Time: 20:16
 */
public class Producer {

    //交换机名称
    static final String EXCHAGE = "topic_exchange";
    //队列名称
    static final String QUEUE_1 = "topic_queue_1";
    //队列名称
    static final String QUEUE_2 = "topic_queue_2";

    public static void main(String[] args) {
        try(
                // 创建连接
                Connection connection = ConnectionUtil.getConnection();
                // 创建频道
                Channel channel = connection.createChannel()
        ){
            /**
                声明交换机
                exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments)
               参数:
                1. exchange:交换机名称
                2. type:交换机类型
                    DIRECT("direct"),:定向
                    FANOUT("fanout"),:扇形(广播),发送消息到每一个与之绑定队列。
                    TOPIC("topic"),通配符的方式
                    HEADERS("headers");参数匹配

                3. durable:是否持久化
                4. autoDelete:自动删除
                5. internal:内部使用。 一般false
                6. arguments:参数
             */
            channel.exchangeDeclare(EXCHAGE, BuiltinExchangeType.TOPIC,true,false,false,null);

            // 声明(创建)队列
            /**
             * 参数1:队列名称
             * 参数2:是否定义持久化队列
             * 参数3:是否独占本次连接
             * 参数4:是否在不使用的时候自动删除队列
             * 参数5:队列其它参数
             */
            channel.queueDeclare(QUEUE_1,true,false,false,null);
            channel.queueDeclare(QUEUE_2,true,false,false,null);

            /**
             * 队列绑定交换机
            queueBind(String queue, String exchange, String routingKey)
            参数:
                1. queue:队列名称
                2. exchange:交换机名称
                3. routingKey:路由键,绑定规则
                    如果交换机的类型为fanout ,routingKey设置为""
             */
            channel.queueBind(QUEUE_1,EXCHAGE,"#.error");
            channel.queueBind(QUEUE_2,EXCHAGE,"user.*");
            channel.queueBind(QUEUE_2,EXCHAGE,"*.*");

            // 要发送的信息
            String mes = "rabitmq---info";
            /**
             * 参数1:交换机名称,如果没有指定则使用默认Default Exchage
             * 参数2:路由key,简单模式可以传递队列名称
             * 参数3:消息其它属性
             * 参数4:消息内容
             */
            channel.basicPublish(EXCHAGE, "user.info", null, mes.getBytes());
            System.out.println("已发送消息:" + mes);

        }catch (Exception e){
            e.printStackTrace();
        }
    }

}

2.消费者:one::
package com.glj.rabitmq.topic;

import com.glj.rabitmq.util.ConnectionUtil;
import com.rabbitmq.client.*;

import java.io.IOException;

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: GuoLiangJun
 * @Date: 2021/6/20
 * @Time: 20:44
 */
public class Consumer1 {

    //交换机名称
    static final String EXCHAGE = "topic_exchange";
    //队列名称
    static final String QUEUE_1 = "topic_queue_1";

    public static void main(String[] args) throws Exception {

        // 创建连接
        Connection connection = ConnectionUtil.getConnection();
        // 创建频道
        Channel channel = connection.createChannel();

        /**
         声明交换机
         exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments)
         参数:
         1. exchange:交换机名称
         2. type:交换机类型
         DIRECT("direct"),:定向
         FANOUT("fanout"),:扇形(广播),发送消息到每一个与之绑定队列。
         TOPIC("topic"),通配符的方式
         HEADERS("headers");参数匹配

         3. durable:是否持久化
         4. autoDelete:自动删除
         5. internal:内部使用。 一般false
         6. arguments:参数
         */
        channel.exchangeDeclare(EXCHAGE, BuiltinExchangeType.TOPIC,true,false,false,null);

        DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
            @Override
            /**
             回调方法,当介绍消息后,会自动执行该方法
             * consumerTag 消息者标签,在channel.basicConsume时候可以指定
             * envelope 消息包的内容,可从中获取消息id,消息routingkey,交换机,消息和重传标志(收到消息失败后是否需要重新发送)
             * properties 属性信息
             * body 消息
             */
            public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                //收到的消息
                System.out.println("消费者1-接收到的消息为:" + new String(body, "utf-8"));

            }
        };
        //监听消息
        /**
         * 参数1:队列名称
         * 参数2:是否自动确认,设置为true为表示消息接收到自动向mq回复接收到了,mq接收到回复会删除消息,设置为false则需要手动确认
         * 参数3:消息接收到后回调
         */
        channel.basicConsume(QUEUE_1,true,defaultConsumer);

        //不关闭资源,应该一直监听消息
        //channel.close();
        //connection.close();
    }

}

3.消费者:two::
package com.glj.rabitmq.topic;

import com.glj.rabitmq.util.ConnectionUtil;
import com.rabbitmq.client.*;

import java.io.IOException;

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: GuoLiangJun
 * @Date: 2021/6/20
 * @Time: 20:44
 */
public class Consumer2 {

    //交换机名称
    static final String EXCHAGE = "topic_exchange";
    //队列名称
    static final String QUEUE_1 = "topic_queue_2";

    public static void main(String[] args) throws Exception {

        // 创建连接
        Connection connection = ConnectionUtil.getConnection();
        // 创建频道
        Channel channel = connection.createChannel();

        /**
         声明交换机
         exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments)
         参数:
         1. exchange:交换机名称
         2. type:交换机类型
         DIRECT("direct"),:定向
         FANOUT("fanout"),:扇形(广播),发送消息到每一个与之绑定队列。
         TOPIC("topic"),通配符的方式
         HEADERS("headers");参数匹配

         3. durable:是否持久化
         4. autoDelete:自动删除
         5. internal:内部使用。 一般false
         6. arguments:参数
         */
        channel.exchangeDeclare(EXCHAGE, BuiltinExchangeType.TOPIC,true,false,false,null);

        DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
            @Override
            /**
             回调方法,当介绍消息后,会自动执行该方法
             * consumerTag 消息者标签,在channel.basicConsume时候可以指定
             * envelope 消息包的内容,可从中获取消息id,消息routingkey,交换机,消息和重传标志(收到消息失败后是否需要重新发送)
             * properties 属性信息
             * body 消息
             */
            public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                //收到的消息
                System.out.println("消费者1-接收到的消息为:" + new String(body, "utf-8"));

            }
        };
        //监听消息
        /**
         * 参数1:队列名称
         * 参数2:是否自动确认,设置为true为表示消息接收到自动向mq回复接收到了,mq接收到回复会删除消息,设置为false则需要手动确认
         * 参数3:消息接收到后回调
         */
        channel.basicConsume(QUEUE_1,true,defaultConsumer);

        //不关闭资源,应该一直监听消息
        //channel.close();
        //connection.close();
    }

}

小结

Topic主题模式可以实现 Publish/Subscribe发布与订阅模式Routing路由模式 的功能;只是Topic在配置routing key 的时候可以使用通配符,显得更加灵活。

9.6. 模式总结

RabbitMQ工作模式:

1、简单模式 HelloWorld

一个生产者、一个消费者,不需要设置交换机(使用默认的交换机)

2、工作队列模式 Work Queue

一个生产者、多个消费者(竞争关系),不需要设置交换机(使用默认的交换机)

3、发布订阅模式 Publish/subscribe

需要设置类型为fanout的交换机,并且交换机和队列进行绑定,当发送消息到交换机后,交换机会将消息发送到绑定的队列

4、路由模式 Routing

需要设置类型为direct的交换机,交换机和队列进行绑定,并且指定routing key,当发送消息到交换机后,交换机会根据routing key将消息发送到对应的队列

5、通配符模式 Topic

需要设置类型为topic的交换机,交换机和队列进行绑定,并且指定通配符方式的routing key,当发送消息到交换机后,交换机会根据routing key将消息发送到对应的队列

10.Spring Boot整合RabbitMQ

在Spring项目中,可以使用Spring-Rabbit去操作RabbitMQ
https://github.com/spring-projects/spring-amqp

尤其是在spring boot项目中只需要引入对应的amqp启动器依赖即可,方便的使用RabbitTemplate发送消息,使用注解接收消息。

一般在开发过程中

生产者工程:

  1. application.properties文件配置RabbitMQ相关信息;
  2. 在生产者工程中编写配置类,用于创建交换机和队列,并进行绑定

  3. 注入RabbitTemplate对象,通过RabbitTemplate对象发送消息到交换机

消费者工程:

  1. application.properties文件配置RabbitMQ相关信息

  2. 创建消息处理类,用于接收队列中的消息并进行处理

10.1 搭建生产者工程

创建生产者工程rabitmq-producer-boot

image-20210622001415320

pom.xml 引入:

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

10.1.1配置RabbitMQ

配置文件application.properties
spring.rabbitmq.host=127.0.0.1
spring.rabbitmq.port=5672
spring.rabbitmq.username=hcxtuser
spring.rabbitmq.password=hcxtuser123!@#
spring.rabbitmq.virtual-host=/hcxtuser
绑定交换机和队列

创建RabbitMQ队列与交换机绑定的配置类com.glj.rabbitmq.config.RabbitMQConfig

package com.glj.rabbitmq.config;

import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: GuoLiangJun
 * @Date: 2021/6/22
 * @Time: 0:02
 */
@Configuration
public class RabbitMQConfig {

    //交换机名称
    public static final String EXCHANGE_NAME = "boot_topic_exchange";
    //队列名称
    public static final String QUEUE_NAME = "boot_queue";

    //声明交换机
    @Bean("bootExchange")
    public Exchange bootExchange(){
        return ExchangeBuilder.topicExchange(EXCHANGE_NAME).durable(true).build();
    }

    //声明队列
    @Bean("bootQueue")
    public Queue bootQueue(){
        return QueueBuilder.durable(QUEUE_NAME).build();
    }

    //绑定队列和交换机
    @Bean
    public Binding bootQueueExchange(@Qualifier("bootQueue") Queue queue,
                                     @Qualifier("bootExchange") Exchange exchange){
        return BindingBuilder.bind(queue).to(exchange).with("boot.#").noargs();
    }

}

测试.
package com.glj;

import com.glj.rabbitmq.config.RabbitMQConfig;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

@SpringBootTest
@RunWith(SpringRunner.class)
public class RabitmqProducerBootApplicationTests {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    public void testSend() {
        String msg = "此消息:boot.rabbitmq";
        rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE_NAME,"boot.rabbitmq",msg);
    }

}

image-20210622001742704

10.2 搭建消费者工程

image-20210622002020872

pom.xml 和application.properties与生产者一致...

10.2.1配置RabbitMQ

消息监听处理类

编写消息监听器com.glj.rabbitmq.listener.RabbitMQListener

package com.glj.rabbitmq.listener;

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: GuoLiangJun
 * @Date: 2021/6/22
 * @Time: 0:23
 */

import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class RabbitMQListener {

    @RabbitListener(queues = "boot_queue")
    public void ListenQueue(Message message){
        System.out.println("消费者接收到的消息为:" + new String(message.getBody()));
    }

}

11.RabbitMQ 高级特性

11.1 消息的可靠投递

在使用 RabbitMQ 的时候,作为消息发送方希望杜绝任何消息丢失或者投递失败场景。RabbitMQ 为我们提供了两种方式用来控制消息的投递可靠性模式。

rabbitmq 整个消息投递的路径为:

producer--->rabbitmq broker--->exchange--->queue--->consumer

我们将利用这两个 callback 控制消息的可靠性投递

confirm 确认模式

消息从 producer 到 exchange 则会返回一个 confirmCallback

在spring boot 生产者中继续咱们的操作.

1.确认模式的开启

在配置文件application.properties 新增 spring.rabbitmq.publisher-confirms=true

2.在rabbitTemplate 定义ConfirmCallBack回调函数

package com.glj.rabbitmq.config;

import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: GuoLiangJun
 * @Date: 2021/6/22
 * @Time: 0:02
 */
@Configuration
public class RabbitMQConfigWithConfirm {

    public static final String CONFIRM_EXCHANGE_NAME = "confirm_exchange";
    public static final String CONFIRM_QUEUE_NAME = "confirm_queue";


    //确认模式

    @Bean("confirmExchange")
    public Exchange confirmExchange(){
        return new TopicExchange(CONFIRM_EXCHANGE_NAME,true,false,null);
    }

    @Bean("confirmQueue")
    public Queue confirmQueue(){
        return new Queue(CONFIRM_QUEUE_NAME,true,false,false);
    }

    @Bean
    public Binding confirmQueueExchange(@Qualifier("confirmQueue") Queue queue,
                                        @Qualifier("confirmExchange") Exchange exchange){
        return BindingBuilder.bind(queue).to(exchange).with("confirm").noargs();
    }

}

测试类中:

@Test
public void testConfirm(){
    // 定义回调
    rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
        /**
             *
             * @param correlationData 相关配置信息
             * @param ack exchange交换机 是否成功收到消息
             * @param cause 失败原因
             */
        @Override
        public void confirm(CorrelationData correlationData, boolean ack, String cause) {
            System.out.println("confirm方法被执行了..."+ack);
            if (!ack){
                //接受成功
                System.out.println("接受"+cause);
            }else{
                //失败
                System.out.println("失败原因:"+cause);
            }
        }
    });
    String msg = "此消息:confirm 方法测试";
    rabbitTemplate.convertAndSend(RabbitMQConfigWithConfirm.CONFIRM_EXCHANGE_NAME,"confirm",msg);
}

return 退回模式

消息从 exchange-->queue 投递失败则会返回一个 returnCallback 。

在spring boot 生产者中继续咱们的操作.

1.确认模式的开启

在配置文件application.properties 新增 spring.rabbitmq.publisher-returns=true

2.在rabbitTemplate 定义ReturnCallBack回调函数

@Test
public void testReturn(){
    //设置交换机处理失败消息的模式
    rabbitTemplate.setMandatory(true);

    //设置回调
    rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
        /**
             *
             * @param message 消息对象
             * @param replyCode 返回错误码
             * @param replyText 返回错误信息
             * @param exchange 交换机
             * @param routingKey 路由
             */
        @Override
        public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
            System.out.println("return 执行了....");
            System.out.println("message:"+message);
            System.out.println("replyCode:"+replyCode);
            System.out.println("replyText:"+replyText);
            System.out.println("exchange:"+exchange);
            System.out.println("routingKey:"+routingKey);
        }
    });
    String msg = "此消息:confirm-return 方法测试";
    rabbitTemplate.convertAndSend("confirm_exchange","confirm111",msg);
}

消息的可靠投递小结

设置ConnectionFactory的publisher-confirms="true" 开启 确认模式。

使用rabbitTemplate.setConfirmCallback设置回调函数。当消息发送到exchange后回调confirm方法。在方法中判断ack,如果为true,则发送成功,如果为false,则发送失败,需要处理。

设置ConnectionFactory的publisher-returns="true" 开启 退回模式。

使用rabbitTemplate.setReturnCallback设置退回函数,当消息从exchange路由到queue失败后,如果设置了rabbitTemplate.setMandatory(true)参数,则会将消息退回给producer。并执行回调函数returnedMessage。

在RabbitMQ中也提供了事务机制,但是性能较差,此处不做讲解。使用channel下列方法,完成事务控制:

  1. txSelect(), 用于将当前channel设置成transaction模式
  2. txCommit(),用于提交事务
  3. txRollback(),用于回滚事务

Consumer Ack

ack指Acknowledge,确认。 表示消费端收到消息后的确认方式。

有三种确认方式:

  1. 自动确认:acknowledge="none"
  2. 手动确认:acknowledge="manual"
  3. 根据异常情况确认:acknowledge="auto"

其中自动确认是指,当消息一旦被Consumer接收到,则自动确认收到,并将相应 message 从 RabbitMQ 的消息缓存中移除。但是在实际业务处理中,很可能消息接收到,业务处理出现异常,那么该消息就会丢失。如果设置了手动确认方式,则需要在业务处理成功后,调用channel.basicAck(),手动签收,如果出现异常,则调用channel.basicNack()方法,让其自动重新发送消息。

消费端限流

Ø配置 prefetch属性设置消费端一次拉取多少消息

Ø消费端的确认模式一定为手动确认。acknowledge="manual"

TTL

TTL 全称 Time To Live(存活时间/过期时间)。

当消息到达存活时间后,还没有被消费,会被自动清除。

RabbitMQ可以对消息设置过期时间,也可以对整个队列(Queue)设置过期时间。

实现:

设置队列过期时间使用参数:x-message-ttl,单位:ms(毫秒),会对整个队列消息统一过期。

设置消息过期时间使用参数:expiration。单位:ms(毫秒),当该消息在队列头部时(消费时),会单独判断这一消息是否过期。

如果两者都进行了设置,以时间短的为准。

死信队列

死信队列,英文缩写:DLX 。Dead Letter Exchange(死信交换机),当消息成为Dead message后,可以被重新发送到另一个交换机,这个交换机就是DLX。

image-20210624220505093

消息成为死信的三种情况:

  1. 队列消息长度到达限制;

  2. 消费者拒接消费消息,basicNack/basicReject,并且不把消息重新放入原目标队列,requeue=false;

  3. 原队列存在消息过期设置,消息到达超时时间未被消费;

队列绑定死信交换机:

给队列设置参数: x-dead-letter-exchange 和 x-dead-letter-routing-key

image-20210624220537822

死信队列小结

1.死信交换机和死信队列和普通的没有区别

2.当消息成为死信后,如果该队列绑定了死信交换机,则消息会被死信交换机重新路由到死信队列

3.消息成为死信的三种情况:

  1. 队列消息长度到达限制;

  2. 消费者拒接消费消息,并且不重回队列;

  3. 原队列存在消息过期设置,消息到达超时时间未被消费;

延迟队列

延迟队列,即消息进入队列后不会立即被消费,只有到达指定时间后,才会被消费。

使用:TTL+死信队列 组合实现延迟队列的效果。

image-20210624220811711

延迟队列小结
  1. 延迟队列 指消息进入队列后,可以被延迟一定时间,再进行消费。

  2. RabbitMQ没有提供延迟队列功能,但是可以使用 : TTL + DLX 来实现延迟队列效果。

发表评论

您的电子邮箱地址不会被公开。 必填项已用*标注