1.概述
RabbitMQ作为消息队列，有6种队列模型，分别在不同的场景进行使用，分别是Hello World，Work queues，Publish/Subscribe，Routing，Topics，RPC。
下面就分别对几个模型进行讲述。
2.Hello World
这个模型也叫直连模型，从这个名字来看就知道它的原理很简单，是一个线性的且没有分支的模型。
生产者生产消息，把消息给队列，队列在把消息给消费者进行消费。
代码实现
工具类
public class RabbitMQUtils {public static Connection getConnection() throws IOException, TimeoutException {//创建一个连接工厂ConnectionFactory factory = new ConnectionFactory();factory.setHost("192.168.153.132");//设置rabbitmq的主机地址factory.setPort(5672);//设置rabbitmq的端口号factory.setUsername("admin");//设置用户名factory.setPassword("admin");//设置密码factory.setVirtualHost("/abc");//设置虚拟主机//根据连接工厂获取一个连接对象Connection connection = factory.newConnection();return connection;}
}生产者
/*** 基于HelloWorld消息模型的消息生产者*/
public class MessageProvider1 {public static void main(String[] args) throws IOException, TimeoutException {Connection connection= RabbitMQUtils.getConnection();//根据连接对象创建一个Channel(信道)Channel channel = connection.createChannel();/*** 根据Channel声明一个队列* 参数1：队列的名称* 参数2：描述队列是否持久化（true 讲队列以文件的形式保存在硬盘上，下一次启动mq服务还可以看到该队列）* 参数3：是否独占队列（true 只有当前会话才能使用该队列）* 参数4：队列是否会自动删除* 参数5：对队列进行的额外设置*/channel.queueDeclare("hello", true, false, false, null);/*** 发布消息* 参数1：交换机的名称* 参数2：队列的名称* 参数3：消息进行额外设置 MessageProperties.PERSISTENT_TEXT_PLAIN消息持久化* 参数4：消息主体，以字节数组的方式进行消息的发送*/channel.basicPublish("","hello", MessageProperties.PERSISTENT_TEXT_PLAIN,"hello rabbitmq".getBytes());//关闭资源channel.close();connection.close();}
}消费者部分
/*** 基于HelloWorld消息模型的消息消费者*/
public class MessageConsumer1 {public static void main(String[] args) throws IOException, TimeoutException {Connection connection= RabbitMQUtils.getConnection();Channel channel = connection.createChannel();//声明一个队列channel.queueDeclare("hello", true, false, false, null);/*** 消息的消费* 参数1：队列的名称* 参数2：是否开启自动确认机制 true 开启* 参数3：处理队列里面消息的回调函数*/channel.basicConsume("hello",true,new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {String msg = new String(body);System.out.println(msg);}});}
}这时候先启动消费者等待生产者的消息，然后再开启生产者，那么消费者就会接收到消息，这里是”Hello rabbit“
这里的队列名称要自己再RabbitMQ的管理界面进行创建
在这个界面可以进行手动创建
3.Work queues
它与第一种模型不同的是，消息队列不再是单一的讲消息传送给一个消费者，而是可以传送给多个消费者。它就解决了生产者生产消息的速率大于消费者消费消息的速率，使消息不断地堆积在队列中。
同样的它的模型也很简单，也是生产消息给队列，队列再将消息分给队列。
代码实现
生产者
/*** 消息的生产者 -- 工作队列实现*/
public class MessageProvider2 {public static void main(String[] args) throws Exception {Connection connection = RabbitMQUtils.getConnection();// 创建信道Channel channel = connection.createChannel();// 声明队列channel.queueDeclare("work", false, false, false, null);// 发送消息for (int i = 1; i <= 10; i++) {String message = "hello,SpringCloud" + i;channel.basicPublish("", "work", null, message.getBytes());}// 关闭资源channel.close();connection.close();}
}消费者
public class MessageConsumer_q2 {public static void main(String[] args) throws Exception {Connection connection = RabbitMQUtils.getConnection();final Channel channel = connection.createChannel();//声明队列channel.queueDeclare("work",false,false,false,null);//消费消息 参数2 false 关闭消息的自动确认机制channel.basicConsume("work",true,new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope,AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("消息是:" + new String(body));}});}
}消费者的代码都一样的，就不重复写了。
但现在要思考一个问题，就是我这里的代码是开启了消息的自动确认机制，就是意思是队列会将消息按照顺序平均分配给消费者，也就是消费者会先再队列中确认消息（确认意味着消息从队列中删除），这样就会出现一个问题，有的消费者消费得快，有得消费者消费得慢，这样得结果显然不是我们想要的，我们肯定希望能者多劳，也就是按需分配，而不是按量平均分配 。还有一个问题，就是消费者一下子确认这么多消息然后慢慢消费，万一某个消费者在消费的时候挂掉了，这些消息都已经不在队列中了，那不就直接导致消息丢失了。
因此我们需要对消息进行手动确认而不是自动确认
public class MessageConsumer_q1 {public static void main(String[] args) throws Exception {Connection connection = RabbitMQUtils.getConnection();final Channel channel = connection.createChannel();//声明队列channel.queueDeclare("work", false, false, false, null);//消费消息 第二个参数改成false，即关闭自动确认channel.basicConsume("work", false, new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {// 开启手动确认 不再同时确认多条消息channel.basicAck(envelope.getDeliveryTag(), false);System.out.println("消息是:" + new String(body));}});}
}4.fanout
这个模型就相对复杂点，多了一个交换机。
广播类型的交换机可以和任意类型的队列进行配对。
每个消费者有自己的队列。
每个队列都要绑定到交换机.
生产者发送的消息，只能发送到交换机，交换机来决定要发给哪个队列，生产者无法决定。
交换机把消息发送给绑定过的所有队列 队列的消费者都能拿到消息。
实现一条消息被多个消费者消费
代码实现
生产者
public class MessageProvider3 {public static void main(String[] args) throws IOException, TimeoutException {Connection connection = RabbitMQUtils.getConnection();Channel channel = connection.createChannel();/*** 声明交换机* 参数1：交换机的名称* 参数2：交换机的类型 fanout广播类型的交换机（可以和任何队列进行绑定）*/channel.exchangeDeclare("exchange_fanout","fanout");/*** 发布消息* 参数1：交换机的名称* 参数2；路由key 由于是广播类型的交换机，所以不用定义路由key* 参数3：对消息进行的额外设置* 参数4：消息主体*/channel.basicPublish("exchange_fanout","",null,"hello rabbitmq".getBytes());}
}消费者
public class MessageConsumer3 {public static void main(String[] args) throws Exception {Connection connection = RabbitMQUtils.getConnection();Channel channel = connection.createChannel();//声明交换机channel.exchangeDeclare("exchange_fanout", "fanout");//声明队列channel.queueDeclare("q1", false, false, false, null);//将交换机和队列进行绑定 参数1: 队列的名称 参数2: 交换机的名称 参数3: 路由keychannel.queueBind("q1", "exchange_fanout", "");//消费消息channel.basicConsume("q1", true, new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println(new String(body));}});}
}一条消息会被所有消费者消费
5.Direct
在前面的模型中，一条消息被所有的消费者消费，但是在某些特定的场景下，我们希望不同的消息被不同的队列消费
生产者向交换机发送消息时，会指定一个routing key
交换机在接收到消息时会根据这个key来对它绑定的队列的routing key进行匹配，匹配的才会发送过去。
代码实现
生产者
/*** 消息的生产者 -- direct消息模型*/
public class MessageProvider {public static void main(String[] args) throws Exception {// 获取与RabbitMQ的连接Connection connection = RabbitMQUtils.getConnection();// 创建信道Channel channel = connection.createChannel();// 声明交换机，名称为exchange_direct，类型为directchannel.exchangeDeclare("exchange_direct", "direct");// 发布消息，发送到exchange_direct交换机，routingKey为add，消息内容为"hello, this is direct message"channel.basicPublish("exchange_direct", "add", null, "hello, this is direct message".getBytes());// 关闭信道channel.close();// 关闭连接connection.close();}
}消费者1
/*** 消息的消费者1 -- direct类型*/
public class ConsumerProvider_1 {public static void main(String[] args) throws Exception {// 获取与RabbitMQ的连接Connection connection = RabbitMQUtils.getConnection();// 创建信道Channel channel = connection.createChannel();// 声明交换机，名称为exchange_direct，类型为directchannel.exchangeDeclare("exchange_direct", "direct");// 声明队列，名称为queue_directchannel.queueDeclare("queue_direct", false, false, false, null);// 使用routingKey绑定队列和交换机，分别绑定add、update的routingKeychannel.queueBind("queue_direct", "exchange_direct", "add");channel.queueBind("queue_direct", "exchange_direct", "update");// 消费消息，设置为自动确认channel.basicConsume("queue_direct", true, new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {// 打印接收到的消息内容System.out.println(new String(body));}});}
}消费者2
/*** 消息的消费者2 -- direct类型*/
public class ConsumerProvider_2 {public static void main(String[] args) throws Exception {// 获取与RabbitMQ的连接Connection connection = RabbitMQUtils.getConnection();// 创建信道Channel channel = connection.createChannel();// 声明交换机，名称为exchange_direct，类型为directchannel.exchangeDeclare("exchange_direct", "direct");// 声明队列，名称为queue_direct1channel.queueDeclare("queue_direct1", false, false, false, null);// 使用routingKey绑定队列和交换机，分别绑定update、delete的routingKeychannel.queueBind("queue_direct1", "exchange_direct", "update");channel.queueBind("queue_direct1", "exchange_direct", "delete");// 消费消息，设置为自动确认channel.basicConsume("queue_direct1", true, new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {// 打印接收到的消息内容System.out.println(new String(body));}});}
}消费者1的队列与交换机绑定的routing key分别是 add、update
消费者2的队列与交换机绑定的routing key分别是update、delete
示例代码中的生产者生产的消息的routing key是add，那么只有消费者1能接收到消息。
6.Topics
这个模型与Direct的原理是一样的，差异就在于它的routing key是有模糊匹配的机制。
但是它的routing key一般是由多个单词拼在一起，中间由”.“进行隔开，例如item.insert
这里的通配符有两种
分别是*和#
*可以匹配一个单词
#可以匹配一个或多个单词
代码演示
生产者
public class MessageProvider5 {public static void main(String[] args) throws Exception {Connection connection = RabbitMQUtils.getConnection();Channel channel = connection.createChannel();//声明topic类型的交换机channel.exchangeDeclare("topic_exchange", "topic");//发布消息channel.basicPublish("topic_exchange","user.product.add",null,"hello,this is topic message".getBytes());//关闭资源channel.close();connection.close();}
}消费者1
public class MessageConsumer5_1 {public static void main(String[] args) throws Exception {Connection connection = RabbitMQUtils.getConnection();Channel channel = connection.createChannel();//声明交换机channel.exchangeDeclare("topic_exchange", "topic");//声明队列channel.queueDeclare("topic_q1", false, false, false, null);//将队列和交换机进行绑定channel.queueBind("topic_q1", "topic_exchange", "user.*");//消费消息channel.basicConsume("topic_q1", true, new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println(new String(body));}});}
}消费者2
public class MessageConsumer5_2 {public static void main(String[] args) throws Exception {Connection connection = RabbitMQUtils.getConnection();Channel channel = connection.createChannel();//声明交换机channel.exchangeDeclare("topic_exchange", "topic");//声明队列channel.queueDeclare("topic_q2", false, false, false, null);//将队列和交换机进行绑定channel.queueBind("topic_q2", "topic_exchange", "user.#");//消费消息channel.basicConsume("topic_q2", true, new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println(new String(body));}});}
}7.RPC(了解）
- 客户端：
- 与 RabbitMQ 建立连接并创建信道。
- 声明一个用于接收响应的回调队列。
- 生成一个唯一的
correlationId，用于匹配请求和响应。 - 发送请求消息到 RPC 队列，同时设置
replyTo为回调队列名称，correlationId为生成的唯一标识。 - 从回调队列接收响应消息，并根据
correlationId进行匹配。
- 服务端：
- 与 RabbitMQ 建立连接并创建信道。
- 声明 RPC 队列。
- 从 RPC 队列接收请求消息。
- 处理请求消息，生成响应结果。
- 将响应消息发送到
replyTo指定的回调队列，并带上相同的correlationId。