RabbitMQ02_简单模式Publish/Subscribe发布与订阅模式Routing路由模式Topics通配符模式Work模式-轮询公平
Posted 所得皆惊喜
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了RabbitMQ02_简单模式Publish/Subscribe发布与订阅模式Routing路由模式Topics通配符模式Work模式-轮询公平相关的知识，希望对你有一定的参考价值。
文章目录
①. Simple简单模式
- ①. 简单模式分析如下:
生产者:发送消息到RabbitMQ队列(simple_queue)
消费者:接收RabbitMQ队列消息
- ②. 导入对于的依赖、工具类的抽取
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>5.6.0</version>
</dependency>
public class ConnectionUtil
public static Connection getConnection() throws IOException, TimeoutException
//1. 创建连接工厂(设置RabbitMQ的连接参数)；
ConnectionFactory connectionFactory = new ConnectionFactory();
//主机；默认localhost
connectionFactory.setHost("139.198.169.136");
//连接端口；默认5672
connectionFactory.setPort(5672);
//虚拟主机；默认/
connectionFactory.setVirtualHost("/myvitrualhost");
//用户名；默认guest
connectionFactory.setUsername("tang");
//密码；默认guest
connectionFactory.setPassword("9602111022yxTZ@");
//2. 创建连接；
return connectionFactory.newConnection();
- ③. 编写生产者
注意:设置虚拟机,一个mq服务可以设置多个虚拟机,每个虚拟机就相当于一个独立的mq
public class Producer
static final String QUEUE_NAME = "simple_queue";
public static void main(String[] args) throws Exception
//1. 创建连接工厂(设置RabbitMQ的连接参数)；
ConnectionFactory connectionFactory = new ConnectionFactory();
//主机；默认localhost
connectionFactory.setHost("139.198.169.136");
//连接端口；默认5672
connectionFactory.setPort(5672);
//虚拟主机；默认/
connectionFactory.setVirtualHost("/myvitrualhost");
//用户名；默认guest
connectionFactory.setUsername("tang");
//密码；默认guest
connectionFactory.setPassword("9602111022yxTZ@");
//2. 创建连接；
Connection connection = connectionFactory.newConnection("简单模式生产者");
//3. 创建频道；
Channel channel = connection.createChannel();
//4. 声明队列:如果队列在mq中没有则要创建
/**
* 参数1:队列名称:
* 参数2:是否定义持久化队列(消息会持久化保存在服务器上)
* 所谓持久化消息是否存盘,如果false,非持久化
* 参数3:是否独占本连接,队列只允许在该连接中访问,如果连接关闭后队列自动删除,
* (如果将此参数设置为true可用于临时队列的创建)
* 参数4:是否在不使用的时候队列自动删除,如果将此参数和参数3设置为true就可以实现
* 临时队列[队列不用了就自动删除]
* 参数5:参数,可以设置一个队列的扩展参数,比如:可设置存活时间
*/
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
//5. 发送消息；
String message = "你好！小兔纸。";
/**
* 参数1:交换机名称；如果没有则指定空字符串(表示使用默认的交换机)
* 参数2:路由key,简单模式中可以使用队列名称
* 交换机根据路由key来将消息转发到指定的队列,如果使用默认的交换机,
* routingKey要设置为队列的名称
* 参数3:消息其它属性
* 参数4:消息内容
*/
channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
System.out.println("已发送消息:" + message);
//6. 关闭资源
channel.close();
connection.close();
- ④. 编写消费者
需要持续监听队列消息,所以不要关闭资源
public class Consumer
public static void main(String[] args) throws Exception
//1. 创建连接工厂；
//2. 创建连接；(抽取一个获取连接的工具类)
Connection connection = ConnectionUtil.getConnection();
//3. 创建频道；
Channel channel = connection.createChannel();
//4. 声明队列；
/**
* 参数1:队列名称
* 参数2:是否定义持久化队列(消息会持久化保存在服务器上)
* 参数3:是否独占本连接
* 参数4:是否在不使用的时候队列自动删除
* 参数5:其它参数
*/
//channel.queueDeclare(Producer1.QUEUE_NAME, true, false, false, null);
//5. 创建消费者(接收消息并处理消息)；
DefaultConsumer defaultConsumer = new DefaultConsumer(channel)
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException
//路由key
System.out.println("路由key为:" + envelope.getRoutingKey());
//交换机
System.out.println("交换机为:" + envelope.getExchange());
//消息id
//消息id,mq在channel中用来标识消息的id,可用来确认消息已接收
System.out.println("消息id为:" + envelope.getDeliveryTag());
//接收到的消息
System.out.println("接收到的消息为:" + new String(body, "utf-8"));
;
//6. 监听队列
/**
* 参数1:队列名
* 参数2:是否要自动确认；设置为true表示消息接收到自动向MQ回复接收到了,MQ则会将消息从队列中删除；
* 如果设置为false则需要手动确认
* 参数3:消费方法,当消费者接收到消息要执行的方法
*/
channel.basicConsume(Producer.QUEUE_NAME, true, defaultConsumer);
- ⑤. 总结如下
- ⑥. 图形化操作简单模式
②. Publish/Subscribe发布与订阅模式
-
①. 应用场景(用户通知,当用户充值成功或转账完成系统通知用户,通知方式有短信、邮件多种方法)
-
②. 模式说明
发布与订阅模式:一个消息可以被多个消费者接收。一个消费者对于的队列,该队列只能被一个消费者监听。使用了订阅模式中交换机类型为:广播
- ③. 生产者
/**
* 发布与订阅模式:发送消息
*/
public class Producer
//交换机名称
static final String FANOUT_EXCHANGE = "fanout_exchange";
//队列名称
static final String FANOUT_QUEUE_1 = "fanout_queue_1";
//队列名称
static final String FANOUT_QUEUE_2 = "fanout_queue_2";
public static void main(String[] args) throws Exception
//1. 创建连接；
Connection connection = ConnectionUtil.getConnection();
//2. 创建频道；
Channel channel = connection.createChannel();
//3. 声明交换机；参数1:交换机名称,参数2:交换机类型(fanout,direct,topic)
channel.exchangeDeclare(FANOUT_EXCHANGE, BuiltinExchangeType.FANOUT);
//4. 声明队列；
/**
* 参数1:队列名称
* 参数2:是否定义持久化队列(消息会持久化保存在服务器上)
* 参数3:是否独占本连接
* 参数4:是否在不使用的时候队列自动删除
* 参数5:其它参数
*/
channel.queueDeclare(FANOUT_QUEUE_1, true, false, false, null);
channel.queueDeclare(FANOUT_QUEUE_2, true, false, false, null);
//5. 队列绑定到交换机；参数1:队列名称,参数2:交换机名称,参数3:路由key
channel.queueBind(FANOUT_QUEUE_1, FANOUT_EXCHANGE, "");
channel.queueBind(FANOUT_QUEUE_2, FANOUT_EXCHANGE, "");
//6. 发送消息；
for(int i = 1; i<=10; i++)
String message = "你好！小兔纸。发布订阅模式 --- " + i;
/**
* 参数1:交换机名称；如果没有则指定空字符串(表示使用默认的交换机)
* 参数2:路由key,简单模式中可以使用队列名称
* 参数3:消息其它属性
* 参数4:消息内容
*/
channel.basicPublish(FANOUT_EXCHANGE, "", null, message.getBytes());
System.out.println("已发送消息:" + message);
//6. 关闭资源
channel.close();
connection.close();
- ④. 消费者
/**
* 发布与订阅模式；消费者接收消息
*/
public class Consumer
private static Runnable runnable=()->
try
//1. 创建连接；(抽取一个获取连接的工具类)
Connection connection = ConnectionUtil.getConnection();
//2. 创建频道；
Channel channel = connection.createChannel();
//3.申明队列queue存储消息
/*
* 如果队列不存在,则会创建
* Rabbitmq不允许创建两个相同的队列名称,否则会报错。
* @params1: queue 队列的名称
* @params2: durable 队列是否持久化
* @params3: exclusive 是否排他,即是否私有的,如果为true,会对当前队列加锁,其他的通道不能访问,并且连接自动关闭
* @params4: autoDelete 是否自动删除,当最后一个消费者断开连接之后是否自动删除消息。
* @params5: arguments 可以设置队列附加参数,设置队列的有效期,消息的最大长度,队列的消息生命周期等等。
* */
// 这里如果queue已经被创建过一次了,可以不需要定义
// channel.queueDeclare("queue1", false, false, false, null);
//6. 创建消费者(接收消息并处理消息)；
DefaultConsumer defaultConsumer = new DefaultConsumer(channel)
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException
//路由key
System.out.println("路由key为:" + envelope.getRoutingKey());
//交换机
System.out.println("交换机为:" + envelope.getExchange());
//消息id
System.out.println("消息id为:" + envelope.getDeliveryTag());
//接收到的消息
System.out.println("消费者1 --- 接收到的消息为:" + new String(body, "utf-8"));
;
//6. 监听队列
/**
* 参数1:队列名
* 参数2:是否要自动确认；设置为true表示消息接收到自动向MQ回复接收到了,MQ则会将消息从队列中删除；
* 如果设置为false则需要手动确认
* 参数3:消费者
*/
channel.basicConsume(Thread.currentThread().getName(), true, defaultConsumer);
catch (Exception e)
e.getLocalizedMessage();
;
public static void main(String[] args)
new Thread(runnable, Producer.FANOUT_QUEUE_1).start();
new Thread(runnable, Producer.FANOUT_QUEUE_2).start();
- ⑤. 测试:
- 启动所有消费者,然后使用生产者发送消息;在每个消费者对应的控制台可以查看到生产者发送的所有消息;达到广播的效果
- 在执行完测试代码后,其实到RabbitMQ的管理后台找到Exchanges选项卡,点击fanout_exchange的交换机,可以查看到如下的绑定
③. Routing路由模式
-
①. Routing模式和Publish/subscibe有啥区别？
Routing模式要求队列在绑定交换机时要指定routingkey,消息会转发到符合routingkey的队列 -
②. Routing路由模式
- P:生产者,向Exchange发送消息,发送消息时,会指定一个routing key
- X:Exchange(交换机),接收生产者的消息,然后把消息递交给与routing key完全匹配的队列
- C1:消费者,其所在队列指定了需要routing key为error的消息
- C2:消费者,其所在队列指定了需要routing key为info、error、warning的消息
- ③. 生产者
public class Producer
//交换机名称
static final String DIRECT_EXCHANGE = "direct_exchange";
//队列名称
static final String DIRECT_QUEUE_INSERT = "direct_queue_insert";
//队列名称
static final String DIRECT_QUEUE_UPDATE = "direct_queue_update";
public static void main(String[] args) throws Exception
//1. 创建连接；
Connection connection = ConnectionUtil.getConnection();
//2. 创建频道；
Channel channel = connection.createChannel();
//3. 声明交换机；参数1:交换机名称,参数2:交换机类型(fanout,direct,topic)
channel.exchangeDeclare(DIRECT_EXCHANGE, BuiltinExchangeType.DIRECT);
//4. 声明队列；
/**
* 参数1:队列名称
* 参数2:是否定义持久化队列(消息会持久化保存在服务器上)
* 参数3:是否独占本连接
* 参数4:是否在不使用的时候队列自动删除
* 参数5:其它参数
*/
channel.queueDeclare(DIRECT_QUEUE_INSERT, true, false, false, null);
channel.queueDeclare(DIRECT_QUEUE_UPDATE, true, false, false, null);
//5. 队列绑定到交换机；参数1:队列名称,参数2:交换机名称,参数3:路由key
channel.queueBind(DIRECT_QUEUE_INSERT, DIRECT_EXCHANGE, "insert");
channel.queueBind(DIRECT_QUEUE_INSERT, DIRECT_EXCHANGE, "delete");
channel.queueBind(DIRECT_QUEUE_UPDATE, DIRECT_EXCHANGE, "update");
//6. 发送消息；
String message = "你好！小兔纸。路由模式;routing key为insert";
/**
* 参数1:交换机名称；如果没有则指定空字符串(表示使用默认的交换机)
* 参数2:路由key,简单模式中可以使用队列名称
* 参数3:消息其它属性
* 参数4:消息内容
*/
channel.basicPublish(DIRECT_EXCHANGE, "insert", null, message.getBytes());
System.out.println("已发送消息:" + message);
message = "你好！小兔纸。路由模式 ；routing key 为 update ";
/**
* 参数1:交换机名称；如果没有则指定空字符串(表示使用默认的交换机)
* 参数2:路由key,简单模式中可以使用队列名称
* 参数3:消息其它属性
* 参数4:消息内容
*/
channel.basicPublish(DIRECT_EXCHANGE, "update", null, message.getBytes());
message = "你好！小兔纸。路由模式 ；routing key 为 delete ";
/**
* 参数1:交换机名称；如果没有则指定空字符串(表示使用默认的交换机)
* 参数2:路由key,简单模式中可以使用队列名称
* 参数3:消息其它属性
* 参数4:消息内容
*/
channel.basicPublish(DIRECT_EXCHANGE, "delete", null, message.getBytes());
System.out.println("已发送消息:" + message);
//6. 关闭资源
channel.close();
connection.close();
- ④. 消费者
/**
* 路由模式；消费者接收消息
*/
public class Consumer
static Runnable runnable=()->
try
//1. 创建连接；(抽取一个获取连接的工具类)
Connection connection = ConnectionUtil.getConnection();
//2. 创建频道；
Channel channel = connection.createChannel();
//3. 创建消费者(接收消息并处理消息)；
DefaultConsumer defaultConsumer = new DefaultConsumer(channel)
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException
//路由key
System.out.println("路由key为:" + envelope.getRoutingKey());
//交换机
System.out.println("交换机为:" + envelope.getExchange());
//消息id
System.out.println("消息id为:" + envelope.getDeliveryTag());
//接收到的消息
System.out.println("消费者1 --- 接收到的消息为:" + new String(body, "utf-8"));
;
//4. 监听队列
/**
* 参数1:队列名
* 参数2:是否要自动确认；设置为true表示消息接收到自动向MQ回复接收到了,MQ则会将消息从队列中删除；
* 如果设置为false则需要手动确认
* 参数3:消费者
*/
String name = Thread.currentThread().getName();
System.out.println("线程的名字:"+name);
channel.basicConsume(name, true, defaultConsumer);
catch (以上是关于RabbitMQ02_简单模式Publish/Subscribe发布与订阅模式Routing路由模式Topics通配符模式Work模式-轮询公平的主要内容，如果未能解决你的问题，请参考以下文章
学相伴狂神说 RabbitMQ笔记（简单使用RabbitMQ）
学相伴狂神说 RabbitMQ笔记（简单使用RabbitMQ）