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){
@Override
public 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){
@Override
public 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){
@Override
public 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: 路由key
channel.queueBind("q1", "exchange_fanout", "");
//消费消息
channel.basicConsume("q1", 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));
}
});
}
}一条消息会被所有消费者消费
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，类型为direct
channel.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，类型为direct
channel.exchangeDeclare("exchange_direct", "direct");
// 声明队列，名称为queue_direct
channel.queueDeclare("queue_direct", false, false, false, null);
// 使用routingKey绑定队列和交换机，分别绑定add、update的routingKey
channel.queueBind("queue_direct", "exchange_direct", "add");
channel.queueBind("queue_direct", "exchange_direct", "update");
// 消费消息，设置为自动确认
channel.basicConsume("queue_direct", 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));
}
});
}
}消费者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，类型为direct
channel.exchangeDeclare("exchange_direct", "direct");
// 声明队列，名称为queue_direct1
channel.queueDeclare("queue_direct1", false, false, false, null);
// 使用routingKey绑定队列和交换机，分别绑定update、delete的routingKey
channel.queueBind("queue_direct1", "exchange_direct", "update");
channel.queueBind("queue_direct1", "exchange_direct", "delete");
// 消费消息，设置为自动确认
channel.basicConsume("queue_direct1", 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的队列与交换机绑定的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){
@Override
public 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){
@Override
public 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。
