【RabbitMQ（day2）】默认（直连）交换机的应用

这篇博客是以下资料学后的总结：
不良人的RabbitMQ的教学视频
官方启动教程
RabbitMQ中文文档

一、第一种模型（Hello World）

在上图的模型中，有以下概念：

• P：生产者，也就是要发送消息的程序；
• C：消费者：消息的接受者，会一直等待消息的到来。
• queue：消息队列，图中红色部分。类似于一个邮箱，可以缓存消息；生产者向其中投递消息，消费者从其中取出消息。

1. 开发生产者

		// 创建连接mq的连接工厂对象ConnectionFactory connectionFactory = new ConnectionFactory();// 设置连接Rabbitmq的主机connectionFactory.setHost("192.168.248.135");// 设置端口号connectionFactory.setPort(5672);// 设置连接那个虚拟主机connectionFactory.setVirtualHost("/ems");// 设置用户名和密码connectionFactory.setUsername("ems");connectionFactory.setPassword("123456");// 获取连接对象Connection connection = connectionFactory.newConnection();// 获取连接中的通道对象Channel channel = connection.createChannel();// 通过通道绑定对应的消息队列// 参数1：队列的名称  如果队列不存在会自动创建// 参数2：用来定义队列特性是否需要持久化，true:持久化队列，false即不持久化// 参数3：exclusive 是否独占队列// 参数4：是否在消费完成后自动删除队列// 参数5：额外参数// 这个不加是没关系的，只是表示我的Rabbitmq中是有hello消息队列的，消费者产生的
channel.queueDeclare("hello",false,false,false,null);// 发布消息// 参数1：交换机名称；参数2：路由键名称；参数3：传递消息额外设置；参数4：消息的具体内容channel.basicPublish("","hello",null,"hello rabbitmq".getBytes());channel.close();connection.close();

2. 消费者开始消费

 		// 创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();// 设置主机和端口connectionFactory.setHost("192.168.248.135");connectionFactory.setPort(5672);// 设置虚拟主机connectionFactory.setVirtualHost("/ems");// 设置用户名和密码connectionFactory.setUsername("ems");connectionFactory.setPassword("123456");// 创建连接对象Connection connection = connectionFactory.newConnection();// 创建通道Channel channel = connection.createChannel();// 通道绑定对象channel.queueDeclare("hello",false,false,false,null);// 消费消息// 参数1：消费哪个队列的消息  队列名称// 参数2：开始消息的自动确认机制// 参数3：消费消息时的回调接口String hello = channel.basicConsume("hello", true, new DefaultConsumer(channel){// 最后一个参数：消息队列中取出的消息@Overridepublic void handleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body) throws IOException {System.out.println("转化成对应的字符串： " + new String(body));System.out.println("============");}});/*channel.close();connection.close();*/

这里需要注意，由于这里是多线程下生产、消费消息，所以在消费时不应该提前关闭通道，不然无法监听到队列中的数据。

下面是证明，看看各线程的名称就知道了。

需注意：Junit5之前是不支持多线程的。

3. 参数说明

channel.queueDeclare("hello",true,false,true,null);
"参数1"：用来声明通道对应的队列；
"参数2"：用来指定是否持久化队列
"参数3"：用来指定是否独占队列，一般为false
"参数4"：用来指定是否自动删除队列
"参数5"：对队列的额外配置参数1：交换机名称；参数2：路由键名称；参数3：传递消息额外设置；参数4：消息的具体内容
channel.basicPublish("","hello", MessageProperties.PERSISTENT_TEXT_PLAIN,"I love you~".getBytes());

二、第二种模型（work queue）

Work Queues，也被称为（Task Queues 任务模型）。当消息处理比较耗时的时候，可能生产消息的速度会远远大于消息的消费速度。长此以往，消息就会堆积越来越多，无法及时处理。此时就可以使用 work 模型：让多个消费者绑定到一个队列，共同消费队列中的消息。队列中的消息一旦消费，就会消失，因此任务是不会被重复执行的。

角色：

• Sender：生产者：任务的发布者
• Consumer：消费者，领取任务并且完成任务

生产者代码

        // 获取连接对象Connection conn = RabbitMQUtils.getConnection();// 获取通道对象Channel channel = conn.createChannel();// 通过通道声明队列channel.queueDeclare("work", true, false, false, null);for (int i = 0; i < 10; ++i) {// 生产消息channel.basicPublish("", "work", null, ("(" + i + ")Hello worke queue~").getBytes());}// 关闭资源RabbitMQUtils.closeConnectionAndChanel(conn, channel);

消费者代码

Runnable myRunnable = new Runnable() {@Overridepublic void run() {Connection conn = RabbitMQUtils.getConnection();try {final int[] cnt = new int[1];cnt[0] = 0;Channel channel = conn.createChannel();channel.queueDeclare("work", true, false, false, null);System.out.println("当前线程：" + Thread.currentThread().getName());channel.basicConsume("work", true, new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body) throws IOException {System.out.println("消费者-" + consumerTag + "：" + new String(body));System.out.println("================================================");cnt[0]++;}});System.out.println(Thread.currentThread().getName() + ":" + cnt[0]);} catch (Exception e) {e.printStackTrace();}}};Thread work1 = new Thread(myRunnable, "work-001");Thread work2 = new Thread(myRunnable, "work-002");Thread work3 = new Thread(myRunnable, "work-003");work1.start();work2.start();work3.start();

测试结果

总结：默认情况下，RabbitMQ将按顺序将每个消息发送给下一个使用者。平均而言，每个消费者都会收到相同数量的消息。这种分发消息的方式称为轮询。如下图所示：

下面是官方给的

By default, RabbitMQ will send each message to the next consumer, in sequence. On average every consumer will get the same number of messages. This way of distributing messages is called round-robin. Try this out with three or more workers.

自动确认机制的后果和公平分配

channel.basicConsume("work", true, new DefaultConsumer(channel)
// 这里的第二个参数是指是否默认提交

上面 work queue 实现会出现两个问题：

问题一：这里将 autoAcked 参数值设置为了 true，即消费者收到消息队列调度的消息后不管有没有消费成功都立即返回 ACK 确认，消息队列只顾着轮询分配去了。这个时候的话会引发一个问题：当消费者突然宕机了，那还没处理的消息就不会被处理，即消失了。比如一个消费者被分配到了五个消息，但是只处理了三个就嘎了，那剩下的俩个就处理不了了。

问题二：在两个worker的情况下，当所有奇数消息都很重，偶数消息都很轻时，一个worker将一直很忙，而另一个几乎不做任何工作。但是RabbitMQ对此一无所知，仍然会均匀地分发消息。我们应该遵循能者多劳，充分利用资源，但轮询方式总是这么的不合我们的胃口。

解决方案

• 首先得将自动提交设置为 false，手动提交就好了；

• 每一次给空闲发消费者一个消息，即设置 prefetchCount = 1，这样的话不会让能者出现不工作，懒者一堆事没做的情况。当消费者死亡（即通道关闭、连接被关闭、或者TCP连接丢失等情况）还没有发送ACK，那有其他消费者在线的话，消息队列会将消息迅速交付给另一个消费者，从而确保消息没有丢失。

  具体解决方案的伪代码如下：

// 配置每一次只能执行一个小希
channel.basicQos(1);
// 关闭手动提交
channel.basicConsume("work", false, new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body) throws IOException {System.out.println("消费者-" + consumerTag + "：" + new String(body));// 参数1：确认队列中哪些具体消息 参数2：是否开启多个消息同时确认channel.basicAck(envelope.getDeliveryTag(), false);}
});

三、阐述默认交换机

可以在 RabbitMQ中文文档-默认交换机 去了解更多AMQP协议的一些内容。

默认交换机的本质是直连交换机，当你添加一个队列的时候，这个队列第一反应就是绑定默认交换机，而绑定（binding）的路由键名称和队列名称是一致的。

上面两种模型（RabbitMQ官方教程阐述的）Hello World模型和Work Queues模型，在官方教程中没有指出使用了交换机，但是本质都是绑定了默认交换机的，也就是直连交换机，它也是支持多消费者的负载均衡的。

首先必须知道的是：使用默认交换机时，队列是在消费者端创建的（可以说是用户本身吧），而不是生产者去创建的。当生产者发送一条消息到 RabbitMQ 时，RabbitMQ 会根据消息的路由键（在使用默认交换机的情况下，路由键即为队列名称）来查找是否已经存在该队列，如果队列不存在，则会丢弃该消息。

从 basicPublish 方法的第二个参数为路由键名称和 basicConsume 方法的第一个参数为队列名称也是可以看出来的。

总的来说就是生产者不需要关心队列的创建，这是消费者需要声明指定的，默认交换机会绑定声明的消息队列的，所以生产者该发发，创建的任务就不用管了。