【RabbitMQ】常用消息模型详解

AMQP协议的回顾

在RabbitMQ中有生产者、消费者的概念，生产者先与我们的RabbitMQServer建立连接，建立完连接之后，它会把消息通过连接中通道的形式去传递我们的消息。每一个生产者会对应一个专门的虚拟主机。

在我们做项目的时候，RabbitMQ希望我们每一个项目具有单独的虚拟主机，这样我们多个应用在操作同一个RabbitMQServer的时候互不影响，所以这里的虚拟机有点像关系型数据库中的库概念。

我们在访问虚拟主机的时候是需要权限的，如果需要访问到某一个具体的虚拟主机，我们需要将虚拟主机与用户进行绑定。

比如RabbitMQ默认为我们提供的guest账户，他是可以访问所有的虚拟主机的，具有至高无上的权限。在我们实际的生产环境中我们一般是一个项目访问一个虚拟主机，或者说是一个业务访问一个虚拟主机，在访问的时候我们一般为一个虚拟主机绑定特定的用户。

当我们的生产者通过通道将消息放入到虚拟机之中，因为RabbitMQ存在许多的消息模型，所以这里不一定会把消息放入到交换机之中。也就是说当生产者将消息传递给交换机或者队列之后，他的任务就告一段落了。

这个时候我们的生产者和消费者是完全解耦的，我们不需要关心生产者到底有没有运行，我只关心消费者监听的队列里面有没有对应的消息即可。

消费者在消费消息的时候也需要去连接到我们RabbitMQServer以及虚拟主机，我们才能消费到对应主机中的消息队列里面的数据。

RabbitMQ支持的消息模型

最新的版本有第七种消息模型：消息确认模型

第一种模型(直连)

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

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

首先我们先创建一个新用户/ems，然后将一个虚拟主机与其绑定，然后给他添加超级用户权限：

注意：
用户名必须以/开头

开发生产者

public class Provider {//生产消息@Testpublic void testSendMessage() throws IOException, TimeoutException {//创建连接mq的连接工厂对象ConnectionFactory connectionFactory = new ConnectionFactory();//设置连接rabbitmqserver主机connectionFactory.setHost("10.15.0.9");//设置端口号connectionFactory.setPort(5672);//设置连接那个虚拟主机connectionFactory.setVirtualHost("/ems");//设置访问虚拟主机的用户名和密码connectionFactory.setUsername("ems");connectionFactory.setPassword("123");//获取连接对象Connection connection = connectionFactory.newConnection();//获取连接中通道Channel channel = connection.createChannel();//通道绑定对应消息队列//参数1:  队列名称 如果队列不存在自动创建//参数2:  用来定义队列特性是否要持久化 true 持久化队列   false 不持久化//参数3:  exclusive 是否独占队列  true 独占队列   false  不独占//参数4:  autoDelete: 是否在消费完成后自动删除队列  true 自动删除  false 不自动删除//参数5:  额外附加参数channel.queueDeclare("hello",true,false,false,null);//发布消息//参数1: 交换机名称 参数2:队列名称  参数3:传递消息额外设置  参数4:消息的具体内容channel.basicPublish("","hello", null,"hello rabbitmq".getBytes());channel.close();connection.close();}
}

开发消费者

public class Customer {public static void main(String[] args) throws IOException, TimeoutException {//创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("10.15.0.9");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/ems");connectionFactory.setUsername("ems");connectionFactory.setPassword("123");//创建连接对象Connection connection = connectionFactory.newConnection();*///创建通道Channel channel = connection.createChannel();//通道绑定对象channel.queueDeclare("hello",true,false,false,null);//消费消息//参数1: 消费那个队列的消息 队列名称//参数2: 开始消息的自动确认机制//参数3: 消费时的回调接口channel.basicConsume("hello",true,new DefaultConsumer(channel){@Override //最后一个参数: 消息队列中取出的消息public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("===================================="+new String(body));}});}}

注意：
在使用Junit测试的时候，他是不支持多线程模型的。如果我们使用@Test去运行的话，他没法让我们的消费者去监听(运行完之后直接就杀死了该进程，不会处于监听状态)，所以这里我们要换成一个main函数。
生产者则不需要注意这一点，因为它生产完消息就完事了

我们发现生产者生产完消息之后，会关闭通道和链接，而在消费这里我们并没有这么做。这是因为可能会导致我们的回调函数还没来得及执行，我们的通道就已经关闭。

该模型的特点：
点对点的简单消费模型。
适用于登录、注册场景

生产者、消费者开发优化

我们发现我们在开发生产者、消费者的时候前面的连接部分代码重复冗余，所以我们可以使用一个工具类对其进行封装：

public class RabbitMQUtils {private static ConnectionFactory connectionFactory;private static Properties properties;static{//重量级资源  类加载执行之执行一次connectionFactory = new ConnectionFactory();connectionFactory.setHost("10.15.0.5");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");connectionFactory.setUsername("guest");connectionFactory.setPassword("guest");}//定义提供连接对象的方法public static Connection getConnection() {try {return connectionFactory.newConnection();} catch (Exception e) {e.printStackTrace();}return null;}//关闭通道和关闭连接工具方法public static void closeConnectionAndChanel(Channel channel, Connection conn) {try {if(channel!=null) channel.close();if(conn!=null)   conn.close();} catch (Exception e) {e.printStackTrace();}}}

我们这里使用静态代码块是因为connectionFactory是重量级资源，所以我们决定只在类加载执行时执行一次。

我们这里稍微复习一下java的静态代码块，我们会发现在一些项目源码中经常会见到他。
静态代码块语法格式：

static{}

静态代码块的特点：随着类的加载而执行，而且只执行一次
执行优先级高于非静态的初始化块，它会在类初始化的时候执行一次，执行完成便销毁，它仅能初始化类变量，即static修饰的数据成员。
那么正好我们再来提一下非静态代码块：
非静态代码块语法格式：

{}

执行的时候如果有静态初始化块，先执行静态初始化块再执行非静态初始化块，在每个对象生成时都会被执行一次，它可以初始化类的实例变量。非静态初始化块会在构造函数执行时，在构造函数主体代码执行之前被运行。
执行顺序:
静态代码块----->非静态代码块-------->构造函数

API参数细节

生产者和消费者均有一个方法queueDeclare，就是声明操作的队列：

channel.queueDeclare("hello",true,false,false,null);

• 参数1: 队列名称
  • 如果队列不存在自动创建
• 参数2: 用来定义队列特性是否要持久化
  • true 持久化队列
  • false 不持久化
  • 注意这里说的是队列的持久化
  • 也就是如果开启持久化的话，我们即使重启rabbitmq服务该队列也会存在，因为其内部会把队列从内存写到硬盘中去。当重启完成之后，其又会重新将硬盘中的队列读到内存中去
• 参数3: exclusive 是否独占队列
  • true 独占队列
  • 也就是说队列只能被当前通道所绑定
  • false 不独占
• 参数4: autoDelete: 是否在消费完成后自动删除队列
  • true 自动删除
  • false 不自动删除
  • 这里的自动删除队列是指消费者不再监听占用队列，队列才会消失
• 参数5: 额外附加参数

注意：直连模型下，消费者和生产者的queueDeclare中的参数要保持一致，这样才能保证操作的是同一个队列

生产者：

channel.basicPublish("","hello", MessageProperties.PERSISTENT_TEXT_PLAIN,"hello rabbitmq".getBytes());

• 参数1: 交换机名称 (我们这里没有使用交换机所以没有指定)
• 参数2:队列名称
• 参数3:传递消息额外设置
  • MessageProperties.PERSISTENT_TEXT_PLAIN
  • 我们可以通过此参数设置消息在队列中的持久化
• 参数4:消息的具体内容
  • 这里是以字节的方式进行传输

消费者：

channel.basicConsume("hello",true,new DefaultConsumer(channel){@Override //最后一个参数: 消息队列中取出的消息public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("======"+new String(body));}
});

• 参数1: 消费哪个队列的消息 队列名称
• 参数2: 开始消息的自动确认机制
• 参数3: 消费时的回调接口
  • 这里我们可以传入一个consumer对象，而这个consumer是一个接口，它有一个实现类DefaultConsumer

第二种模型(work quene)

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

角色：

• P：生产者：任务的发布者
• C1：消费者-1，领取任务并且完成任务，假设完成速度较慢
• C2：消费者-2：领取任务并完成任务，假设完成速度快

开发生产者

public class Provider {public static void main(String[] args) throws IOException {//获取连接对象Connection connection = RabbitMQUtils.getConnection();//获取通道对象Channel channel = connection.createChannel();//通过通道声明队列channel.queueDeclare("work", true, false, false, null);for (int i = 1; i <=20; i++) {//生产消息channel.basicPublish("", "work", null, (i + "hello work quene").getBytes());}//关闭资源RabbitMQUtils.closeConnectionAndChanel(channel, connection);}
}

我们这里使用了前面提到的连接工具类

我们运行我们的代码：

这里的Unacked代表未被确认的消息

开发消费者

如果我们对两个消费者不做任何处理：

消费者-1

public class Customer1 {public static void main(String[] args) throws IOException {//获取连接Connection connection = RabbitMQUtils.getConnection();final Channel channel = connection.createChannel();channel.queueDeclare("work",true,false,false,null);//参数1:队列名称  参数2:消息自动确认 true  消费者自动向rabbitmq确认消息消费  false 不会自动确认消息channel.basicConsume("work",false,new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {                System.out.println("消费者-1: "+new String(body));                }});}
}

消费者-2

public class Customer2 {public static void main(String[] args) throws IOException {//获取连接Connection connection = RabbitMQUtils.getConnection();final Channel channel = connection.createChannel();channel.queueDeclare("work",true,false,false,null);//参数1:队列名称  参数2:消息自动确认 true  消费者自动向rabbitmq确认消息消费  false 不会自动确认消息channel.basicConsume("work",false,new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {                System.out.println("消费者-2: "+new String(body));                }});}
}

在这种不做任何处理的情况下，消费者1、消费者2消费的消息都是一致的：

总结:默认情况下，RabbitMQ将按顺序将每个消息发送给下一个使用者。平均而言，每个消费者都会收到相同数量的消息。这种分发消息的方式称为循环。（也就是说平均分配）

而这样的话我们不难想到一个问题：加入我们的消费者1处理的比较慢，消费者2处理的比较快。这就导致消费者1的消息会在队列中造成滞留，消费者2可能已经处理完闲着了。这样的情况下平均分配显然也会影响效率，并且导致消息再队列中的积累。

我们可以模拟一下这个情况，我们在消费者1中添加一个线程睡眠：

这个时候我们运行发现，在消费者2将自己的消息打印完之后，消费者1的消息只打印了一条：

那么能不能用第二种模型实现一种能者多劳的模式呢？

消息自动确认机制

Doing a task can take a few seconds. You may wonder what happens if one of the consumers starts a long task and dies with it only partly done. With our current code, once RabbitMQ delivers a message to the consumer it immediately marks it for deletion. In this case, if you kill a worker we will lose the message it was just processing. We’ll also lose all the messages that were dispatched to this particular worker but were not yet handled.

But we don’t want to lose any tasks. If a worker dies, we’d like the task to be delivered to another worker.
完成一项任务可能需要几秒钟。你可能想知道，如果其中一个消费者开始了一项长任务，但只完成了一部分任务就去世了，会发生什么。使用我们当前的代码，一旦RabbitMQ将消息传递给消费者，它会立即将其标记为删除。在这种情况下，如果你杀死一个消费者，我们将丢失它正在处理的消息。我们还将丢失发送给该特定工作人员但尚未处理的所有消息。
但我们不想失去任何任务。如果一名消费者死亡，我们希望将任务交付给另一名消费者。

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

我们使用能者多劳的模式需要进行两步额外的操作：

• 设置通道一次只能消费一个消息

• 关闭消息的自动确认,开启手动确认消息

Customer1：

public class Customer1 {public static void main(String[] args) throws IOException {//获取连接Connection connection = RabbitMQUtils.getConnection();final Channel channel = connection.createChannel();channel.basicQos(1);//每一次只能消费一个消息channel.queueDeclare("work",true,false,false,null);//参数1:队列名称  参数2:消息自动确认 true  消费者自动向rabbitmq确认消息消费  false 不会自动确认消息channel.basicConsume("work",false,new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {try{Thread.sleep(2000);}catch (Exception e){e.printStackTrace();}System.out.println("消费者-1: "+new String(body));// 参数1:确认队列中那个具体消息 参数2:是否开启多个消息同时确实channel.basicAck(envelope.getDeliveryTag(),false);}});}
}

Customer2：

public class Customer2 {public static void main(String[] args) throws IOException {//获取连接Connection connection = RabbitMQUtils.getConnection();final Channel channel = connection.createChannel();channel.basicQos(1);channel.queueDeclare("work",true,false,false,null);channel.basicConsume("work",false,new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("消费者-2: "+new String(body));//手动确认  参数1:手动确认消息标识  参数2:false 每次确认一个channel.basicAck(envelope.getDeliveryTag(), false);}});    }
}

我们对上面两段代码的一些参数或方法做出解释：

basicQos（）

void basicQos(int prefetchSize, int prefetchCount, boolean global) throws IOException;

参数：

• prefetchSize：消息的大小

• prefetchCount：会告诉RabbitMQ不要同时给一个消费者推送多于N个消息，即一旦有N个消息还没有ack，则该consumer将block掉，直到有消息ack

• global：是否将上面设置应用于channel，简单点说，就是上面限制是channel级别的还是consumer级别

channel.basicAck()

void basicAck(long deliveryTag, boolean multiple) throws IOException;

参数：

• deliveryTag：该消息的index

• multiple：是否批量处理.

  • true:将一次性ack所有小于deliveryTag的消息

envelope.getDeliveryTag()

这个方法是表示消息的唯一标识ID，返回的是一个正整数，是rabbitmq来自增设置的

第三种模型(fanout)

fanout 扇出 也称为广播

在广播模式下，消息发送流程是这样的：

• 可以有多个消费者
• 每个消费者有自己的queue（队列）
  • 我们这里创建的队列是临时的，用完之后就会删除
• 每个队列都要绑定到Exchange（交换机）
• 生产者发送的消息，只能发送到交换机，交换机来决定要发给哪个队列，生产者无法决定。
• 交换机把消息发送给绑定过的所有队列
• 队列的消费者都能拿到消息。实现一条消息被多个消费者消费

使用场景：
比如我们的商品购物车，在我们结算的时候，我们可能会跟多个系统进行交互，订单系统、库存系统等等。这个时候我们的购物车信息会被多条队列给消费。

在fanout模式下，路由的相关配置没有意义，相关参数可以空着

开发生产者

public class Provider {public static void main(String[] args) throws IOException {//获取连接对象Connection connection = RabbitMQUtils.getConnection();Channel channel = connection.createChannel();//将通道声明指定交换机   //参数1: 交换机名称    参数2: 交换机类型  fanout 广播类型channel.exchangeDeclare("logs","fanout");//发送消息channel.basicPublish("logs","",null,"fanout type message".getBytes());//释放资源RabbitMQUtils.closeConnectionAndChanel(channel,connection);}
}

channel.exchangeDeclare("logs","fanout");:

• 参数1: 交换机名称
• 参数2: 交换机类型
  • fanout是广播类型

开发消费者

public class Customer1 {public static void main(String[] args) throws IOException {//获取连接对象Connection connection = RabbitMQUtils.getConnection();Channel channel = connection.createChannel();//通道绑定交换机channel.exchangeDeclare("logs","fanout");//临时队列String queueName = channel.queueDeclare().getQueue();//绑定交换机和队列channel.queueBind(queueName,"logs","");//消费消息channel.basicConsume(queueName,true,new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("消费者1: "+new String(body));}});}
}

其他几个消费者同理

我们创建三个消费者，把他们开启之后，再开启我们的生产者，测试结果如下：

第四种模型(Routing)

第五种模型其实是第四种模型的一个分支，如果我们叫第四种模型为路由的话，那么我们可以说第五种模型是动态路由。第四种模型我们也可以叫做direct模型(直连)

在Fanout模式中，一条消息，会被所有订阅的队列都消费。但是，在某些场景下，我们希望不同的消息被不同的队列消费。这时就要用direct类型的Exchange。

在Routing模型下：

• 队列与交换机的绑定，不能是任意绑定了，而是要指定一个RoutingKey（路由key）
• 消息的发送方在 向 Exchange发送消息时，也必须指定消息的 RoutingKey。
• Exchange不再把消息交给每一个绑定的队列，而是根据消息的Routing Key进行判断，只有队列的Routingkey与消息的 Routing key完全一致，才会接收到消息

流程:

图解：

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

开发生产者

public class Provider {public static void main(String[] args) throws IOException {//获取连接对象Connection connection = RabbitMQUtils.getConnection();//获取连接通道对象Channel channel = connection.createChannel();String exchangeName = "logs_direct";//通过通道声明交换机  参数1:交换机名称  参数2:direct  路由模式channel.exchangeDeclare(exchangeName,"direct");//发送消息String routingkey = "error";channel.basicPublish(exchangeName,routingkey,null,("指定的route key"+key+"的消息").getBytes());//关闭资源RabbitMQUtils.closeConnectionAndChanel(channel,connection);}
}

这个时候我们就已经可以看到我们的交换机了：

开发消费者

我们这里开发两个消费者：

• 消费者1拿到routekey为error的消息
• 消费者2拿到routekey为info、error、warning的消息

public class Customer1 {public static void main(String[] args) throws IOException {Connection connection = RabbitMQUtils.getConnection();Channel channel = connection.createChannel();String exchangeName = "logs_direct";//通道声明交换机以及交换的类型channel.exchangeDeclare(exchangeName,"direct");//创建一个临时队列String queue = channel.queueDeclare().getQueue();//基于route key绑定队列和交换机channel.queueBind(queue,exchangeName,"error");//获取消费的消息channel.basicConsume(queue,true,new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("消费者1: "+ new String(body));}});}
}

public class Customer2 {public static void main(String[] args) throws IOException {Connection connection = RabbitMQUtils.getConnection();Channel channel = connection.createChannel();String exchangeName = "logs_direct";//声明交换机 以及交换机类型 directchannel.exchangeDeclare(exchangeName,"direct");//创建一个临时队列String queue = channel.queueDeclare().getQueue();//临时队列和交换机绑定channel.queueBind(queue,exchangeName,"info");channel.queueBind(queue,exchangeName,"error");channel.queueBind(queue,exchangeName,"warning");//消费消息channel.basicConsume(queue,true,new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("消费者2: "+new String(body));}});}
}

我们测试一下：

测试生产者发送Route key为error的消息时

测试生产者发送Route key为info的消息时

---

第五种模型(Topic)

Topic类型的Exchange与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定Routing key 的时候使用通配符！这种模型Routingkey 一般都是由一个或多个单词组成，多个单词之间以”.”分割，例如： item.insert

• *(star) ：匹配不多不少恰好1个词
• #(hash): 匹配0个或多个词

例如：

audit.#    匹配audit.irs.corporate或者 audit.irs 等
audit.*   只能匹配audit.irs

开发生产者

public class Provider {public static void main(String[] args) throws IOException {//获取连接对象Connection connection = RabbitMQUtils.getConnection();Channel channel = connection.createChannel();//声明交换机以及交换机类型 topicchannel.exchangeDeclare("topics","topic");//发布消息String routekey = "user";channel.basicPublish("topics",routekey,null,("这里是topic动态路由模型,routekey: ["+routekey+"]").getBytes());//关闭资源RabbitMQUtils.closeConnectionAndChanel(channel,connection);}
}

开发消费者

我们还是开发两个消费者：

消费者1Routing Key中使用*通配符方式

public class Customer1 {public static void main(String[] args) throws IOException {//获取连接Connection connection = RabbitMQUtils.getConnection();Channel channel = connection.createChannel();//声明交换机以及交换机类型channel.exchangeDeclare("topics","topic");//创建一个临时队列String queue = channel.queueDeclare().getQueue();//绑定队列和交换机  动态统配符形式route keychannel.queueBind(queue,"topics","user.*");//消费消息channel.basicConsume(queue,true,new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("消费者1: "+ new String(body));}});}
}

消费者2中Routing Key中使用#通配符方式

public class Customer2 {public static void main(String[] args) throws IOException {//获取连接Connection connection = RabbitMQUtils.getConnection();Channel channel = connection.createChannel();//声明交换机以及交换机类型channel.exchangeDeclare("topics","topic");//创建一个临时队列String queue = channel.queueDeclare().getQueue();//绑定队列和交换机  动态统配符形式route keychannel.queueBind(queue,"topics","user.#");//消费消息channel.basicConsume(queue,true,new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("消费者2: "+ new String(body));}});}
}

这个时候我们测试的结果就是：

• 消费者2可以拿到消息
• 消费者1拿不到消息