22-07-18 西安 RabbitMQ (01) RabbitMQ安装、控制控制台界面、RabbitMQ五种模式_rabbitmq控制台

如果你不懂得珍惜自己的注意力，自然会有人替你珍惜，你不管好自己的注意力，它们随时就会被其他人收割、利用...   

---

消息队列MQ

1.MQ特点与作用

消息队列(MQ，Message Queue )是一种应用程序对应用程序的通信方法。应用程序即指一个微服务（项目），我们可以在项目和项目之间架设MQ。
消息队列的主要作用是消除高并发访问高峰，加快网站的响应速度。

既可以保证数据不丢失，也能保证高可用性。

• 高可用即集群部署的时候部分机器宕机可以继续运行。
• 不高可用性，即单节点

先进先出
消息队列就是基础数据结构中的“先进先出”的一种数据机构

消息传递：

指的是程序之间通过消息发送数据进行通信，而不是通过直接调用彼此来通信，直接调用通常是用于诸如远程过程调用的技术。

应用场景：

只有在海量数据、高并发，适用MQ，不要随意的使用。

消息队列的主要的作用

1. 异步处理
2. 解耦服务
3. 流量削峰

---

2.MQ异步处理--发送注册短信

用户注册后（数据入库其实注册功能就实现了）；

为了提供友好的服务：如发送注册邮件和注册短信（关联业务），我们可以采取MQ异步处理

方式一：串行方式

不采取，还不如并行呢，更不如消息队列的异步处理

在项目中，可将一些无需即时返回且耗时的操作提取出来，进行异步处理，而这种异步处理的方式大大的节省了服务器的请求响应时间，从而提高了系统的吞吐量【QPS每秒的响应请求数】

方式二：消息队列
引入消息队列后，用户的响应时间就等于写入数据库的时间+写入消息队列的时间(可以忽略不计。没有必要硬关联，先告诉用户注册成功，正常使用登录以后的功能。

MQ如下图：异步方式处理注册功能，可以加快网站响应速度

---

3.流量削峰

流量削峰一般在秒杀活动中应用广泛【适用场景：极短时间产生大量的流量】

场景举例：数据库只能支撑每秒1000左右的并发写入，并发量再高就容易宕机（如高峰期激增到5000个）

使用MQ解决：MQ把消息保存起来了，数据库可以按照自己的消费能力来消费，比如每秒写入1000个数据，这样慢慢写入数据库，这样就不会卡死数据库了。

但是高峰期产生的数据势必会被积压在MQ中。

在高峰期过后的一段时间内，消费消息的速度还是会维持在1000QPS，直到消费完积压的消息，这就叫做“填谷”

---

4. 名词解释 QPS、PV，UV，PR

QPS即每秒查询率，是对一个特定的查询服务器在规定时间内所处理流量多少的衡量标准。
或者理解：每秒的响应请求数，也即是最大吞吐能力。

计算QPS

QPS = 并发量 / 平均响应时间

原理：每天80%的访问集中在20%的时间里，这20%时间叫做峰值时间。

峰值时间每秒请求数(QPS)

峰值时间每秒请求数(QPS)=( 总PV数 * 80% ) / ( 每天秒数 * 20% )

计算需要的机器数量

需要的机器数量 = 峰值时间每秒QPS / 单台机器的QPS

pV 页面浏览量（点击量)

PV:一个访问者在24小时(0点到24点)内到底看了你网站几个页面

需要强调：同一个人浏览网站同一个页面，不重复计算pv量，点100次也算1次。说白了，pv就是一个访问者打开了你的几个页面。

UV 指访问某个站点或点击某条新闻的不同IP地址的人数。

在同一天内，uv只记录第一次进入网站的具有独立IP的访问者，在同一天内再次访问该网站则不计数.

PR 标识网页的等级

级别从1到10级，10级为满分。PR值越高说明该网页越受欢迎

---

5.Mq方式   【AMQP 和JMS】

AMQP是一种高级消息队列协议（链接协议）
AMQP不从API层进行限定，而是直接定义网络交换的数据格式。

AMQP：​​​​​​​erlang语言实现的， 性能高 稳定性好 支持多种平台(go java c# ...)

JMS即Java消息服务（JavaMessage Service）
JMS基于java实现的消息队列，java定义了消息队列的规范。

实现该规范的消息队列只能由java程序来使用 (消息模型少)

AMQP 和JMS的区别：

• JMS是定义了统一的接口，来对消息操作进行统一；AMQP是通过规定协议来统一数据交互的格式
• JMS限定了必须使用Java语言；AMQP只是协议，不规定实现方式，因此是跨语言的。
• JMS规定了两种消息模式；而AMQP的消息模式更加丰富

主流MQ产品：

• ActiveMQ：基于JMS

• RabbitMQ：基于AMQP协议，erlang语言开发，稳定性好，功能强大，支持多种语言的客户端

• RocketMQ：基于JMS，阿里巴巴产品，目前交由Apache基金会，支持事务消息

• Kafka：分布式消息系统，高吞吐量

---

6. 消息队列产品- 引出RabbitMQ

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

ActiveMQ：基于JMS
Rabbitmq:基于AMQP协议，erlang语言开发，稳定性好
RocketMQ：基于JMS，阿里巴巴产品

---

消息队列RabbitMQ

RabbitMQ是由erlang语言开发，基于AMQP协议实现的消息队列。

Erlang 语言由 Ericson 设计，专门为开发高并发和分布式系统的一种语言，在电信领域使用广泛。

RabbitMQ官方地址：Messaging that just works — RabbitMQ 

1.RabbitMQ提供了7种模式

我们学习;

简单模式，work模式，Publish/Subscribe发布与订阅模式，Routing路由模式，Topics主题模式

官网对应模式介绍：RabbitMQ Tutorials — RabbitMQ 

2.RabbitMQ 中的相关概念

Broker：消息中间件的服务节点；一般情况下可以将一个RabbitMQ Broker看作一台RabbitMQ 服务器。

Virtual host:出于多租户和安全因素设计的，把 AMQP 的基本组件划分到一个虚拟的分组中，类似于网络中的 namespace 概念。当多个不同的用户使用同一个 RabbitMQ server 提供的服务时，可以划分出多个vhost，每个用户在自己的 vhost 创建 exchange／queue 等

Connection：publisher／consumer 和 broker 之间的 TCP 连接

信道Channel：多路复用连接中的一条独立的双向数据流通道。为会话提供物理传输介质。

Exchange：message 到达 broker 的第一站，根据分发规则，匹配查询表中的 routing key，分发消息到queue 中去。

Queue：存储消息的容器，消息最终被送到这里，一个消息可投入一个或多个队列，等待 consumer 取走

Binding：exchange 和 queue 之间的虚拟连接，binding 中可以包含 routing key。Binding 信息被保存到 exchange 中的查询表中，用于 message 的分发依据

客户端Client ：AMQP连接或者会话的发起者。AMQP是非对称的，客户端生产和消费消息，服务器存储和路由这些消息。

端点：AMQP对话的任意一方。一个AMQP连接包括两个端点（一个是客户端，一个是服务器）。

消费者Consumer：一个从消息队列里请求消息的客户端程序。

生产者Producer：一个向交换机发布消息的客户端应用程序

3.RabbitMQ 基础架构图

虚拟主机 Virtual Host【包含交换机和队列】，一个mq server可以包含很多个虚拟主机。

4.RabbitMQ 运转流程

1、connection mq服务（connection只有连接功能）
2、传输信息建立channel【信道】因为是多路复用连接，可以有多个信道channel
3、交换机【Exchange】根据查询表里的 routing key 去分发信息到不同的队列
4、消费端连接服务器，通过信道拿出来信息

---

安装rabbitmq

1、Linux原生安装

需要下载rabbitmq相关安装包如下：

1.下载Erlang的rpm包

RabbitMQ是Erlang语言编写，所以Erlang环境必须要有，注：Erlang环境一定要与RabbitMQ版本匹配：RabbitMQ Erlang Version Requirements — RabbitMQ 

2.下载socat的rpm包

rabbitmq安装依赖于socat，所以需要下载socat。

socat下载地址：http://repo.iotti.biz/CentOS/7/x86_64/socat-1.7.3.2-5.el7.lux.x86_64.rpm

3.下载RabbitMQ的rpm包

RabbitMQ下载地址：Downloading and Installing RabbitMQ — RabbitMQ（根据自身需求及匹配关系，下载对应rpm包）

安装之前一定要，给虚拟机拍快照

然后建立目录“opt/tool/mq”，使用Xftp5传送。这四个文件我当然是老师直接给的了。

rabbitmq的命令都在：
/usr/lib/rabbitmq/bin/ 

安装可视化程序
rabbitmq-plugins enable rabbitmq_management

---

启动   systemctl start rabbitmq-server.service

systemctl status rabbitmq-server.service   

systemctl restart rabbitmq-server.service

systemctl stop rabbitmq-server.service 

查看rabbitmq进程

ps -ef | grep rabbitmq

---

2、Docker安装rabbitmq(推荐！！！)

查询rabbitmq的镜像 

docker search rabbitmq:management

下载镜像：

docker pull rabbitmq:management

创建实例并启动：

• -d: 后台运行容器，并返回容器ID；

• -p: 指定端口映射，格式为：主机(宿主)端口:容器端口

• --name="nginx-lb": 为容器指定一个名称；

docker run -d -p 5672:5672 -p 15672:15672 -p 25672:25672 --name rabbitmq rabbitmq:management

注： 5672 --client通信端口 15672 -- 管理界面ui端口 25672 -- server间内部通信口

---

RabbitMQ管理控制台

1、登录MQ管理控制台

启动后在wndows物理机就可以访问

在web浏览器中输入地址：http://虚拟机ip:15672/

​​​​​​​输入默认账号密码： guest ： guest

guest用户默认不允许远程连接，增加自定义账号

1. 添加管理员账号密码：rabbitmqctl add_user admin admin
2. 分配账号角色：rabbitmqctl set_user_tags admin administrator
3. 修改密码：rabbitmqctl change_password admin 123456
4. 查看用户列表：rabbitmqctl list_users

---

2、管理控制台页面介绍

界面标签页介绍

1. overview：概览
2. connections：无论生产者还是消费者，都需要与RabbitMQ建立连接后才可以完成消息的生产和消费，在这里可以查看连接情况
3. channels：信道，建立连接后，会形成信道，消息的投递获取依赖信道。
4. Exchanges：交换机，用来实现消息的路由
5. Queues：消息队列，消息存放在队列中，等待消费，消费后被移除队列

在管理控制台可以查看端口

5672	rabbitMq的编程语言客户端连接端口
15672	rabbitMq管理界面端口
25672	rabbitMq集群的端口

---

3、RabbitMQ卸载

当然我没有试过，不知道对不对

1. rpm -qa | grep rabbitmq       //查询rpm软件包
2. rpm -e rabbitmq-server       //卸载

linux中软件包安装参考：

22-06-24 西安 linux（01） linux环境搭建、常用命令、vim编辑_￡小羽毛的博客-CSDN博客

---

4、管理控制台添加用户

我创建的用户名和密码都是xiaoyumao,不是图片里的fengge，这个用户名和密码之后会在springboot中的配置文件用到

Tags可选项解释：

1、 超级管理员(administrator)

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

2、 监控者(monitoring)

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

3、 策略制定者(policymaker)

可登录管理控制台， 同时可以对policy进行管理。但无法查看节点的相关信息。

4、 普通管理者(management)

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

5、 其他

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

---

5、 管理控制台创建虚拟主机

虚拟主机：为了分区管理mq中的数据权限，每个虚拟主机都有唯一的路径来标识。

1.mq可以创建账户给它分配操作虚拟主机的权限

2.mq每个虚拟主机创建后多有默认的7个交换机（交换机不保存消息）

虚拟主机：类似于mysql中的database。他们都是以“/”开头

如：我创建的名字就是/xiaoyumao,不是图片里的/fenge

创建虚拟主机Virtual host后效果如下：

​​​​​​​

---

 6、设置用户的权限

1.点击虚拟机/xiaoyumao，进去设置它的用户列表

2.删除了guest用户（我纯属手欠。。），给xiaoyumao用户设置这个虚拟主机的所有权限

3.设置完成后，就能看到虚拟主机/xiaoyumao用户列表多了xiaoyumao这个用户

再退出去，查看用户，也能看到xiaoyumao这个用户可以使用虚拟机/xiaoyumao了

​​​​​​​

---

RabbitMQ 5种模式

1、简单模式

一个生产者、一个消费者，不需要设置交换机（使用默认的交换机）

使用简单模式完成消息传递，不去维护交换机，也是有默认的交换机的

添加依赖 

    <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>

生产端代码

看的出来，在生产端代码 我们是通过信道创建了队列，并把消息发布到了队列里。路由key,简单模式可以传递队列名称

public class Producer {
    public static void main(String[] args) throws Exception {
 
        //创建连接对象
        Connection connection = ConnectionUtil.getConnection();
 
        //创建信道（传输数据）
        Channel channel = connection.createChannel();
 
        /**
         * queue      参数1：队列名称
         * durable    参数2：是否定义持久化队列,当mq重启之后数据还在（给磁盘写数据）
         * exclusive  参数3：是否独占本次连接（是否排它，多个消费者是否可以同时连接同一个队列）
         *            ① 是否独占,只能有一个消费者监听这个队列
         *            ② 当connection关闭时,是否删除队列
         * autoDelete 参数4：是否在不使用的时候自动删除队列,当没有consumer时,自动删除（客户端和队列解除关系是否自动删除队列）
         * arguments  参数5：队列其它参数
         */
        // 通过信道创建队列
        channel.queueDeclare("simple_queue", true, false, false, null);
 
        // 要发送的信息
        String message = "上次这么无语的时候还是在上次！";
        /**
         * 参数1：交换机名称,如果没有指定则使用默认Default Exchage
         * 参数2：路由key,简单模式可以传递队列名称
         * 参数3：配置信息
         * 参数4：消息内容（转为字节）
         */
        //信道发布消息
        channel.basicPublish("", "simple_queue", null, message.getBytes());
        System.out.println("已发送消息：" + message);
 
        // 关闭资源
        channel.close();
        connection.close();
    }
}

在可视化控制台点击队列名。idle是闲置的意思

 可以看见储存在队列里的消息

封装的工具类ConnectionUtil中的代码

public class ConnectionUtil {
    //获取连接对象
    public static Connection getConnection() throws Exception {
        //创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        //主机地址
        connectionFactory.setHost("192.168.2.108");
        //连接端口;默认为 5672
        connectionFactory.setPort(5672);
        //虚拟主机名称;默认为 /
        connectionFactory.setVirtualHost("MyVirtualHost");
        //连接用户名；默认为guest
        connectionFactory.setUsername("admin");
        //连接密码；默认为guest
        connectionFactory.setPassword("123456");
        //创建连接对象
        Connection connection = connectionFactory.newConnection();
        return connection;
    }
}

 消费端代码

autoAck 自动签收消息，如果为true，删除队列中被签收的消息

public class Consumer {
    public static void main(String[] args) throws Exception{
        //消费端获取连接
        Connection connection = ConnectionUtil.getConnection();
        //消费端创建信道
        Channel channel = connection.createChannel();
 
        // 接收消息 consumer可以理解为具体的消费者，在这块获取数据
        DefaultConsumer consumer = new DefaultConsumer(channel){
            /*
               回调方法,当收到消息后,会自动执行该方法
               1. consumerTag：标识
               2. envelope：获取一些信息,交换机,路由key...
               3. properties：配置信息
               4. body：数据
            */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("consumerTag："+consumerTag);
                System.out.println("Exchange："+envelope.getExchange());
                System.out.println("RoutingKey："+envelope.getRoutingKey());
                System.out.println("properties："+properties);
                System.out.println("body："+new String(body));
            }
        };
 
         /*
        basicConsume(String queue, boolean autoAck, Consumer callback)
        参数：
            1. queue：队列名称
            2. autoAck：是否自动确认（自动签收，队列中的消息消费完自动删除） ,类似咱们发短信,发送成功会收到一个确认消息
            3. callback：回调对象（主要目的：获取消费到的消息）
         */
        // 消费者类似一个监听程序,主要是用来监听消息
        channel.basicConsume("simple_queue",true,consumer);
    }
}

其实就是在消费端根据队列名，就能拿到队列里的数据。当然开启了自动签收，则放在mq中的数据就没自动删除了。

---

2、 工作队列模式Work Queues

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

能者多劳：消费者确认消息消费后再消费下一个消息

多个消费端共同消费同一个队列中的消息，目的是为了加快消息的消费速度，提高任务处理的速度。（不能重复消费）

修改生产者的代码，使得生产者给mq中给同一个消息队列“work_queue        ”发送10条数据

// 要发送的信息
//发布消息
for (int i = 1; i <= 10; i++) {
    String body = i+"hello rabbitmq~~~";
    channel.basicPublish("","work_queue",null,body.getBytes());
}

再加一个消费者,2个消费者是轮询的处理了队列里的消息

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

simple和work只有一个队列，所以消息是有序的

---

3、发布订阅模式

需要设置广播(fanout)交换机，并且交换机和队列进行绑定，当发送消息到交换机后，交换机会将消息发送到所有跟它绑定的队列。

订阅模型中，多了一个exchange（交换机）角色，

图解：

• P：生产者，也就是要发送消息的程序，但是不再发送到队列中，而是发给X（交换机）
• C：消费者，消息的接受者，会一直等待消息到来。
• Queue：消息队列，接收消息、缓存消息。
• Exchange：交换机，图中的X。一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。

Exchange的3种类型

Fanout：广播，将消息交给所有绑定到交换机的队列

Direct：定向，把消息交给符合指定routing key 的队列

Topic：通配符（条件更加多元化，条件更具有扩展性），把消息交给符合routing pattern（路由模式） 的队列

消息丢失

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

生产者代码

通过信道创建交换机（指定名字和类型）和队列，并让交换机和队列绑定。最后通过信道发送信息给交换机，Fanout模式下，发送的方法不需要指定“路由key”。

public class Producer {
    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：参数
        */
        String exchangeName = "test_fanout";
        //5. 创建交换机（fanout广播式 只要队列和交换机绑定，就发送消息到该队列）
        channel.exchangeDeclare(exchangeName, BuiltinExchangeType.FANOUT,true,false,false,null);
        //6. 创建队列
        String queue1Name = "test_fanout_queue1";
        String queue2Name = "test_fanout_queue2";
        channel.queueDeclare(queue1Name,true,false,false,null);
        channel.queueDeclare(queue2Name,true,false,false,null);
        //7. 绑定队列和交换机
        /*
        queueBind(String queue, String exchange, String routingKey)
        参数：
            1. queue：队列名称
            2. exchange：交换机名称
            3. routingKey：路由键,绑定条件
                如果交换机的类型为fanout ,routingKey设置为""不指定条件
         */
        channel.queueBind(queue1Name,exchangeName,"");
        channel.queueBind(queue2Name,exchangeName,"");
 
        String body = "废话文学：蝉的翅膀非常薄，到底有多薄呢？薄如蝉翼";
        //8. 发送消息
        channel.basicPublish(exchangeName,"",null,body.getBytes());
 
        //9. 释放资源
        channel.close();
        connection.close();
    }
}

运行结果

交换机与队列进行绑定之后；一个消息可以被多个消费者都收到。（发布-订阅模式）

启动所有消费者，然后使用生产者发送消息；在每个消费者对应的控制台可以查看到生产者发送的所有消息；到达“”广播“”的效果。

到RabbitMQ的管理后台找到Exchanges选项卡，点击 fanout_exchange 的交换机，可以查看到如下的绑定

---

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

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

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

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

---

4、路由模式Routing

需要设置定向（direct的）交换机，交换机和队列进行绑定，并且指定routing key，当发送消息到交换机后，交换机会根据routing key将消息发送到对应的队列（队列保存数据更单一）

路由模式最重要的点

Exchange不再把消息交给每一个绑定的队列，而是根据消息的 Routing Key 进行判断，只有队列的 Routingkey 与消息的  Routing key 完全一致，才会接收到消息

图解定向交换机： 

• P：生产者，向Exchange发送消息，发送消息时，会指定一个routing key。
• X：Exchange（交换机），接收生产者的消息，然后把消息递交给 与routing key完全匹配的队列
• C1：消费者，其所在队列指定了需要routing key 为 error 的消息
• C2：消费者，其所在队列指定了需要routing key 为 info、error、warning 的消息

--------------------------------

路由模式的生产者相比于发布订阅模式，做以下改变

创建交换机的时候，交换机类型变为Direct，队列绑定交换机的的时候指定路由key，信道给交换机发送信息的时候也要指定具体的“路由key”

// 队列1绑定error（暗号）
channel.queueBind(queue1Name,exchangeName,"error");
// 队列2绑定info error warning
channel.queueBind(queue2Name,exchangeName,"info");
channel.queueBind(queue2Name,exchangeName,"error");
channel.queueBind(queue2Name,exchangeName,"warning");
String message = "废话文学：具体情况还得看是什么情况";
//8. 发送消息
channel.basicPublish(exchangeName,"warning",null,message.getBytes());

---

5、通配符模式

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

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

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

通配符规则：
#：匹配一个或多个词（多个包括0个）
*：匹配不多不少恰好1个词

路由key可以使用#(任意多级)和*(一级)通配符 的模型   
生产者将消息发送到交换机并配置一个key，所有的绑定到交换机的队列都需要指定绑定时监听的key
key中可以使用通配符  key满足通配符时才可以接受到消息  

修改交换机的类型为TOPIC，在给队列绑定交换机的时候使用通配符。

• usa.# ，因此凡是以 usa.开头的routing key 都会被匹配到
• #.news ，因此凡是以 .news结尾的 routing key 都会被匹配

// 队列1绑定交换机
channel.queueBind(queue1Name,exchangeName,"usa.#");
// 队列2绑定交换机
channel.queueBind(queue2Name,exchangeName,"#.news");
String message = "废话文学：剥开香蕉后，你会发现一个剥了皮的香蕉";
//8. 发送消息
channel.basicPublish(exchangeName,"usa.news",null,message.getBytes());

RabbitMQ的管理后台可以看到

测试结果不出所料