RabbitMQ 常见的命令1 消息中间件概述
1.1 什么是消息中间件1.2 为什么使用消息队列1.3 AMQP 和 JMS1.4消息队列产品 2 RabbitMQ
2.1 RabbitMQ 安装2.2 RabbitMQ 简单实战 3 RabbitMQ 工作模式
3.1 简单模式3.2work queue 工作队列模式3.3 发布订阅模式3.4 Routing 模式3.5 Topic 通配符模式3.6 模式总结
rabbitmq 常见的 *** 作
RabbitMQ 常见的命令如果没有配置RabbitMq 的相关变量，可以进到rabbitmq 安装目录下的sbin 目录下进行相关 *** 作
rabbitMQ常用的命令
1 消息中间件概述 1.1 什么是消息中间件启动监控管理器：rabbitmq-plugins enable rabbitmq_management
关闭监控管理器：rabbitmq-plugins disable rabbitmq_management
启动rabbitmq：rabbitmq-service start
关闭rabbitmq：rabbitmq-service stop
查看所有的队列：rabbitmqctl list_queues
清除所有的队列：rabbitmqctl reset
关闭应用：rabbitmqctl stop_app
启动应用：rabbitmqctl start_app用户和权限设置
添加用户：rabbitmqctl add_user username password
分配角色：rabbitmqctl set_user_tags username administrator
新增虚拟主机：rabbitmqctl add_vhost vhost_name
将新虚拟主机授权给新用户：rabbitmqctl set_permissions -p vhost_name username “.*” “.*” “.*”(后面三个”*”代表用户拥有配置、写、读全部权限)
MQ全称为(message queue ), 消息队列是用用程序与应用程序之间通信的方法。 多用于分布式之间进行通信。
在远程通信中，通常有两种方式: 1 直接远程调用 如 RPC 框架 （比较有代表性的就是google grpc，它是很多分布式系统的通信基础） 2 借助第三方完成间接通信。比较有代表性的是 Rabbitmq。
1.2 为什么使用消息队列在项目中，可以将一些无需及时返回且耗时的 *** 作提取出来，进行异步处理，而这种异步处理的方式大大的节省了服务器的请求响应时间，从而提高了系统的吞吐量。
消息队列的应用场景如下：
(1) 任务的异步处理。
将不需要同步处理的并且耗时长的 *** 作由消息队列通知消息接收方进行异步处理。提高应用程序的相应时间。
(2) 削峰填谷
在订单系统中，在下单的时候就会往数据库中写数据。但是数据库只能支持每秒1000左右的并发写入。并发量再高就容易宕机。低峰期的时候并发也就100多个，如果遇到双十一这种活动，并发量就会猛增到5000以上，这时候数据库就会卡死。
消息队列这时候就有很大的用处，消息被MQ 保存起来，然后系统就可以按照自己的消费能力来进行消费。比如每秒1000 个数据，这样慢慢写入到数据库中，这样就不会卡死数据库了。
使用了MQ 之后，限制消费的速度为1000，这样一来，高峰期产生的数据势必会被积压在MQ中，高峰就被“削”掉了。但是因为消息积压，在高峰期过后的一段时间内，消费消息的速度还是会维持在1000QPS，直到消费完积压的消息,这就叫做“填谷”。
（3） 应用程序的解耦合
MQ 相当于一个中介，生产方通过MQ 与消费方进行交互，它将应用程序进行解耦合。比如订单系统要调用库存系统，支付系统，这样就会耦合，修改参数会比较麻烦。使用消息队列之后，订单系统给MQ 发送一条消息就算成功了。
1.3 AMQP 和 JMSMQ是消息通信的模型；实现MQ的大致有两种主流方式：AMQP、JMS。
1.2.1. AMQP
AMQP，即 Advanced Message Queuing Protocol（高级消息队列协议），是一个网络协议，是应用层协议的一个开放标准，为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息，并不受客户端/中间件不同产品，不同的开发语言等条件的限制。2006年，AMQP 规范发布。类比HTTP。
AMQP是一种协议，更准确的说是一种binary wire-level protocol（链接协议）。这是其和JMS的本质差别，AMQP不从API层进行限定，而是直接定义网络交换的数据格式。
1.2.2. JMS
JMS即Java消息服务（JavaMessage Service）应用程序接口，是一个Java平台中关于面向消息中间件（MOM）的API，用于在两个应用程序之间，或分布式系统中发送消息，进行异步通信。
1.2.3. AMQP 与 JMS 区别
JMS是定义了统一的接口，来对消息 *** 作进行统一；AMQP是通过规定协议来统一数据交互的格式
JMS限定了必须使用Java语言；AMQP只是协议，不规定实现方式，因此是跨语言的。
JMS规定了两种消息模式；而AMQP的消息模式更加丰富。
现有消息队列产品
ActiveMQ：基于JMSZeroMQ：基于C语言开发RabbitMQ：基于AMQP协议，erlang语言开发，稳定性好RocketMQ：基于JMS，阿里巴巴产品Kafka：类似MQ的产品；分布式消息系统，高吞吐量
2 RabbitMQRabbitMQ 是使用erlang 语言开发的，基于基于AMQP（Advanced Message Queue 高级消息队列协议）协议实现的消息队列，它是一种应用程序之间的通信方法，消息队列在分布式系统开发中应用非常广泛。其整体框架如下。
RabbitMQ 提供了6 种模式，简单模式， work 模式，publish/subscribe 发布与订阅模式，Routing 路由模式，Topics主题模式，RPC远程调用模式。
2.1 RabbitMQ 安装安装rabbitmq 需要安装相应版本的erlang ,其对应的版本可以参考我的博客RabbitMq 与erlang 版本对应 。
安装过程可以参考博客： rabbitmq安装教程。 安装完成之后，进入rabbitMQ 安装目录的sbin 目录。cmd
使用命令
rabbitmq-plugins enable rabbitmq_management
启动监控管理器，其他命令可以参考该博客的第一部分RabbitMQ 的常见的命令。
打开浏览器，地址栏输入http://127.0.0.1:15672 可以看到下面界面说明安装成功。
用户名和密码都可以通过输入guest 进行进入。 可以看到其监控管理界面如下。
最上侧的导航依次是：概览、连接、信道、交换机、队列、用户管理
2.2 RabbitMQ 简单实战1 创建工程，添加依赖。
com.rabbitmq amqp-client5.6.0
下面使用简单模式
创建一个简单的rabbitmq 应用程序
在上图的模型中，有以下概念：
P：生产者，也就是要发送消息的程序C：消费者：消息的接受者，会一直等待消息到来。queue：消息队列，图中红色部分。类似一个邮箱，可以缓存消息；生产者向其中投递消息，消费者从其中取出消息。
2 编写生产者
public class producer_HelloWorld {
static final String QUEUE_NAME = "simple_queue";
public static void main(String[] args) throws IOException, TimeoutException {
//创建连接工厂
ConnectionFactory connectionFactory = new ConnectionFactory();
// 设置一些参数
//主机地址;默认为 localhost
connectionFactory.setHost("localhost");
//连接端口;默认为 5672
connectionFactory.setPort(5672);
//虚拟主机名称;默认为 /
connectionFactory.setVirtualHost("/boshrong");
//连接用户名；默认为guest
connectionFactory.setUsername("guest");
//连接密码；默认为guest
connectionFactory.setPassword("guest");
// 获取对应的连接
//创建连接 Coonecttion
// 根据整体架构图,Connection 中有channel，因此需要创建channel
Connection connection = connectionFactory.newConnection();
// 创建频道 channel
// 简单模式中没有交换机，直接与队列进行交互
Channel channel = connection.createChannel();
// 声明（创建）队列
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
// 要发送的信息
String message = "你好；RabbitMQ！";
channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
System.out.println("已发送消息：" + message);
// 关闭资源
channel.close();
connection.close();
}
}
执行程序之后，可以在管理界面中的 Queues 中看到下面内容。
在虚拟主机 /boshrong 中有一个simple_queue 的队列，其中有一个Ready 的消息。 Connection 中没有相关的资源，因为上面的程序最后将connection 给关闭了。
如果不关闭channel 和 connection 。可以在监控页面看到相应的Connection 和 channel 资源。
3 编写消费者
抽取创建connection的工具类ConnectionUtil
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
public class ConnectionUtils {
public static Connection getConnection() throws Exception {
//创建连接工厂
ConnectionFactory connectionFactory = new ConnectionFactory();
//主机地址;默认为 localhost
connectionFactory.setHost("localhost");
//连接端口;默认为 5672
connectionFactory.setPort(5672);
//虚拟主机名称;默认为 /
connectionFactory.setVirtualHost("/itcast");
//连接用户名；默认为guest
connectionFactory.setUsername("heima");
//连接密码；默认为guest
connectionFactory.setPassword("heima");
//创建连接
return connectionFactory.newConnection();
}
}
编写消费的consumer
import com.rabbitmq.client.*;
import java.io.IOException;
public class Consumer {
static final String QUEUE_NAME = "simple_queue";
public static void main(String[] args) throws Exception {
Connection connection = ConnectionUtil.getConnection();
// 创建频道 注意与producer 中的channel 没有关系
Channel channel = connection.createChannel();
// 声明（创建）队列
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
//创建消费者；并设置消息处理
DefaultConsumer consumer = new DefaultConsumer(channel){
@Override
// handleDelivery 处理收到的消息(回调方法，当收到消息后，会自动执行该方案)
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("接收到的消息为：" + new String(body, "utf-8"));
}
};
//监听消息
channel.basicConsume(QUEUE_NAME, true, consumer);
//不关闭资源，应该一直监听消息
//channel.close();
//connection.close();
}
运行结果如下：
3 RabbitMQ 工作模式RabbitMQ 提供了6 种模式，简单模式， work 模式，publish/subscribe 发布与订阅模式，Routing 路由模式，Topics主题模式，RPC远程调用模式。
3.1 简单模式在上图的模型中，有以下概念：
P：生产者，也就是要发送消息的程序C：消费者：消息的接受者，会一直等待消息到来。queue：消息队列，图中红色部分。类似一个邮箱，可以缓存消息；生产者向其中投递消息，消费者从其中取出消息。
实战可以参考 RabbitMQ 的简单实战
3.2work queue 工作队列模式Work Queues与入门程序的简单模式相比，多了一个或一些消费端，多个消费端共同消费同一个队列中的消息。
应用场景：对于 任务过重或任务较多情况使用工作队列可以提高任务处理的速度。
代码实战：
编写producer 发送30条message
ConnectionUtil 提取出来的配置类
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
public class ConnectionUtil {
public static Connection getConnection() throws Exception {
//创建连接工厂
ConnectionFactory connectionFactory = new ConnectionFactory();
//主机地址;默认为 localhost
connectionFactory.setHost("localhost");
//连接端口;默认为 5672
connectionFactory.setPort(5672);
//虚拟主机名称;默认为 /
connectionFactory.setVirtualHost("/boshrong");
//连接用户名；默认为guest
connectionFactory.setUsername("guest");
//连接密码；默认为guest
connectionFactory.setPassword("guest");
//创建连接
return connectionFactory.newConnection();
}
}
import com.boshrong.rabbitmq.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
public class Producer {
static final String QUEUE_NAME = "work_queue";
public static void main(String[] args) throws Exception {
//创建连接
Connection connection = ConnectionUtil.getConnection();
// 创建频道
Channel channel = connection.createChannel();
// 声明（创建）队列
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
// 发送30 条消息
for (int i = 1; i <= 30; i++) {
// 发送信息
String message = "你好；rabbitmq！work模式--" + i;
channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
System.out.println("已发送消息：" + message);
}
// 关闭资源
channel.close();
connection.close();
}
}
编写消费者: 创建两个消费者 consumer1, consumer2。 其代码一样，只不过只是有两个消费类而已。和消费者1一模一样，复制一份改名Consumer2.java即可。
public class Consumer1 {
public static void main(String[] args) throws Exception {
Connection connection = ConnectionUtil.getConnection();
// 创建频道
Channel channel = connection.createChannel();
// 声明（创建）队列
channel.queueDeclare(Producer.QUEUE_NAME, true, false, false, null);
//一次只能接收并处理一个消息
channel.basicQos(1);
//创建消费者；并设置消息处理
DefaultConsumer consumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
try {
//路由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"));
Thread.sleep(1000);
//确认消息
channel.basicAck(envelope.getDeliveryTag(), false);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
};
//监听消息
channel.basicConsume(Producer.QUEUE_NAME, false, consumer);
}
}
总结: 在一个队列中如果有多个消费者，那么消费者之间对于同一个消息的关系是竞争的关系。
3.3 发布订阅模式之间两种模式，一条消息只能被一个消费者消费，而发布订阅模式中，一条消息可以被多多个消费者进行消费。
发布订阅模式实例图：
前面两种模式只有3个角色：
P ： 生产者： 要发送消息的程序
C： 消费者 ： 消息的接收者，会一直等待消息的到来。
queue: 消息队列， 图中的红色部分。
在发布订阅模式中，多了一个 exchange 角色，而且过程也略有变化：
P： 生产者，也就是要发送消息的程序。但是不再发送到队列中，而是发送到X(交换机)C： 消费者，消息的接收者，会一直等待消息的到来。Queue: 消息队列，接收消息，缓存消息Exchange: 交换机，图中的X，·一方面接收生产者发送过来的消息，另一方面知道如何处理消息。例如递交给某个特别的队列，递交给所有的队列，或者将消息进行遗弃。到底如何 *** 作依靠的是，Exchange 的类型。Exchange的常见类型有下面三种。
Fanout 类型: 广播 将消息交给所有绑定到交换机的消息队列上。Direct 类型: 定向，把消息交给符合指定 routing key 的队列。Topic 类型: 通配符，把消息交给符合routing pattern(路由模式) 的队列。
Exchange (交换机)，只负责转发消息，不具备存储消息的能力。 因此如果没有任何队列与Exchange 进行绑定，或者没有符合路由规则的队列，则消息会丢失。
1 每个消费者监听自己的队列。
2 生产者将消息发送给broker,由交换机将消息转发绑定此交换机的每个队列，每个绑定交换机的队列都接收到消息。
代码实战
声明创建Fanout 类型的交换机, 并声明创建两个队列。 将队列绑定在交换机上，发送消息。
import com.boshrong.rabbitmq.ConnectionUtil;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
public class Producer_PubSub {
//交换机名称
static final String FANOUT_EXCHAGE = "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 {
//创建连接
Connection connection = ConnectionUtil.getConnection();
// 创建频道
Channel channel = connection.createChannel();
channel.exchangeDeclare(FANOUT_EXCHAGE, BuiltinExchangeType.FANOUT);
// 声明（创建）队列
channel.queueDeclare(FANOUT_QUEUE_1, true, false, false, null);
channel.queueDeclare(FANOUT_QUEUE_2, true, false, false, null);
//队列绑定交换机
channel.queueBind(FANOUT_QUEUE_1, FANOUT_EXCHAGE, "");
channel.queueBind(FANOUT_QUEUE_2, FANOUT_EXCHAGE, "");
for (int i = 1; i <= 10; i++) {
// 发送信息
String message = "你好；rabbitmq！发布订阅模式--" + i;
channel.basicPublish(FANOUT_EXCHAGE, "", null, message.getBytes());
System.out.println("已发送消息：" + message);
}
// 关闭资源
channel.close();
connection.close();
}
}
创建两个消费者，进行消费。两个消费者的代码基本上一致，只是指定不同的队列名称而已。
public class Consumer1 {
public static void main(String[] args) throws Exception {
Connection connection = ConnectionUtil.getConnection();
// 创建频道
Channel channel = connection.createChannel();
//声明交换机
channel.exchangeDeclare(Producer_PubSub.FANOUT_EXCHAGE, BuiltinExchangeType.FANOUT);
// 声明（创建）队列
channel.queueDeclare(Producer_PubSub.FANOUT_QUEUE_1, true, false, false, null);
//队列绑定交换机
channel.queueBind(Producer_PubSub.FANOUT_QUEUE_1, Producer_PubSub.FANOUT_EXCHAGE, "");
//创建消费者；并设置消息处理
DefaultConsumer consumer = 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"));
}
};
//监听消息
channel.basicConsume(Producer_PubSub.FANOUT_QUEUE_1, true, consumer);
}
public class Consumer2 {
public static void main(String[] args) throws Exception {
Connection connection = ConnectionUtil.getConnection();
// 创建频道
Channel channel = connection.createChannel();
//声明交换机
channel.exchangeDeclare(Producer_PubSub.FANOUT_EXCHAGE, BuiltinExchangeType.FANOUT);
// 声明（创建）队列
channel.queueDeclare(Producer_PubSub.FANOUT_QUEUE_2, true, false, false, null);
//队列绑定交换机
channel.queueBind(Producer_PubSub.FANOUT_QUEUE_2, Producer_PubSub.FANOUT_EXCHAGE, "");
//创建消费者；并设置消息处理
DefaultConsumer consumer = 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"));
}
};
//监听消息
channel.basicConsume(Producer_PubSub.FANOUT_QUEUE_2, true, consumer);
}
}
运行后可以看到，consumer1, consumer2 都能收到producer 发送过来的十条消息。
启动所有消费者，然后使用生产者发送消息；在每个消费者对应的控制台可以查看到生产者发送的所有消息；到达广播的效果。
小结：
交换机需要与队列进行绑定，绑定之后；一个消息可以被多个消费者都收到。
发布订阅模式与工作队列模式的区别
1、工作队列模式不用定义交换机，而发布/订阅模式需要定义交换机。
2、发布/订阅模式的生产方是面向交换机发送消息，工作队列模式的生产方是面向队列发送消息(底层使用默认交换机)。
3、发布/订阅模式需要设置队列和交换机的绑定，工作队列模式不需要设置，实际上工作队列模式会将队列绑 定到默认的交换机 。
3.4 Routing 模式路由模式特定:
队列与交换机的绑定，不能是任意绑定了，而是要指定一个RouingKey (路由key)消息的发送方在向Exchange 发送消息的时候，也必须指定消息的RoutingKey。Exchange 不再把消息交给每一个绑定的队列，而是根据消息的 RoutingKey 进行判断，只有队列的RoutingKey与消息的RoutingKey 完全一致，才会接收消息。
图解：
P：生产者，向Exchange发送消息，发送消息时，会指定一个routing key。
X：Exchange（交换机），接收生产者的消息，然后把消息递交给 与routing key完全匹配的队列
C1：消费者，其所在队列指定了需要routing key 为 error 的消息
C2：消费者，其所在队列指定了需要routing key 为 info、error、warning 的消息
代码实战：
import com.boshrong.rabbitmq.ConnectionUtil;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
public class producer_routing {
//交换机名称
static final String DIRECT_EXCHAGE = "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 {
//创建连接
Connection connection = ConnectionUtil.getConnection();
// 创建频道
Channel channel = connection.createChannel();
channel.exchangeDeclare(DIRECT_EXCHAGE, BuiltinExchangeType.DIRECT);
// 声明（创建）队列
channel.queueDeclare(DIRECT_QUEUE_INSERT, true, false, false, null);
channel.queueDeclare(DIRECT_QUEUE_UPDATE, true, false, false, null);
//队列绑定交换机,并指定channel 的routing key
channel.queueBind(DIRECT_QUEUE_INSERT, DIRECT_EXCHAGE, "insert");
channel.queueBind(DIRECT_QUEUE_UPDATE, DIRECT_EXCHAGE, "update");
// 发送信息
String message = "新增了商品。路由模式；routing key 为 insert " ;
//
channel.basicPublish(DIRECT_EXCHAGE, "insert", null, message.getBytes());
System.out.println("已发送消息：" + message);
// 发送信息
message = "修改了商品。路由模式；routing key 为 update" ;
channel.basicPublish(DIRECT_EXCHAGE, "update", null, message.getBytes());
System.out.println("已发送消息：" + message);
// 关闭资源
channel.close();
connection.close();
}
运行结果：
创建两个消费者： Consumer1 与 Consumer2
Consumer1 的routing key 为 insert
import com.boshrong.rabbitmq.ConnectionUtil;
import com.rabbitmq.client.*;
import java.io.IOException;
public class Consumer1 {
public static void main(String[] args) throws Exception {
Connection connection = ConnectionUtil.getConnection();
// 创建频道
Channel channel = connection.createChannel();
//声明交换机
channel.exchangeDeclare(producer_routing.DIRECT_EXCHAGE, BuiltinExchangeType.DIRECT);
// 声明（创建）队列
channel.queueDeclare(producer_routing.DIRECT_QUEUE_INSERT, true, false, false, null);
//队列绑定交换机,队列绑定交换机的时候绑定到了insert上
channel.queueBind(producer_routing.DIRECT_QUEUE_INSERT, producer_routing.DIRECT_EXCHAGE, "insert");
//创建消费者；并设置消息处理
DefaultConsumer consumer = 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"));
}
};
//监听消息
channel.basicConsume(producer_routing.DIRECT_QUEUE_INSERT, true, consumer);
}
}
Consumer2 的routing key 为 update
import com.boshrong.rabbitmq.ConnectionUtil;
import com.rabbitmq.client.*;
import java.io.IOException;
public class Consumer2 {
public static void main(String[] args) throws Exception {
Connection connection = ConnectionUtil.getConnection();
// 创建频道
Channel channel = connection.createChannel();
//声明交换机
channel.exchangeDeclare(producer_routing.DIRECT_EXCHAGE, BuiltinExchangeType.DIRECT);
// 声明（创建）队列
channel.queueDeclare(producer_routing.DIRECT_QUEUE_UPDATE, true, false, false, null);
//队列绑定交换机
channel.queueBind(producer_routing.DIRECT_QUEUE_UPDATE, producer_routing.DIRECT_EXCHAGE, "update");
//创建消费者；并设置消息处理
DefaultConsumer consumer = 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("消费者2-接收到的消息为：" + new String(body, "utf-8"));
}
};
//监听消息
channel.basicConsume(producer_routing.DIRECT_QUEUE_UPDATE, true, consumer);
}
Consumer1 运行结果：
Consumer2 运行结果
在管理界面中的 Exchange 选项卡中可以看到，两个队列绑定的routingkey
总结: Routing 模式要求队列在绑定交换机的时候要指定routing key ，消息会转发到符合routing key的队列。
3.5 Topic 通配符模式Topic 类型与Direct 相比，可以根据 Routing Key 把消息路由到不同的队列。只不过Topic 类型Exchange 可以让队列绑定routingkey的时候使用通配符。
RoutingKey 一般都是由一个或多个单词组成。多个单词之间通过“.” 进行分割。例如: item.insert
通配符规则：
#: 匹配零个或多个词
*: 匹配不多不少恰好一个词
举例说明：
item.# ： 能够匹配 item.insert.abc 或者 item.insert
item.*: 能够匹配 item.insert
图解：
红色的Queue : 绑定的是usa.#,因此凡是以usa. 开头的routing key 都能被匹配到。黄色的Queue: 绑定的是 #.news，因此凡是以 .news 结尾的routing key都能被匹配到。
同理，对于其他两个queue 也一样。
代码实战：
下面图就是本次实战要完成的任务。
需求： 所有err 级别的日志存入数据库，所有order 系统的日志存入到数据库，不管什么级别的信息，都打印到控制台。可以把数据库和控制台。当作两个消费者
创建一个producer
import com.boshrong.rabbitmq.ConnectionUtil;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
public class producer_topic {
//交换机名称
static final String TOPIC_EXCHAGE = "topic_exchange";
//队列名称
static final String TOPIC_QUEUE_1 = "topic_queue_1";
//队列名称
static final String TOPIC_QUEUE_2 = "topic_queue_2";
public static void main(String[] args) throws Exception {
// 创建连接
Connection connection = ConnectionUtil.getConnection();
// 创建频道
Channel channel = connection.createChannel();
channel.exchangeDeclare(TOPIC_EXCHAGE, BuiltinExchangeType.TOPIC);
channel.queueDeclare(TOPIC_QUEUE_1, true, false, false, null);
channel.queueDeclare(TOPIC_QUEUE_2, true, false, false, null);
// 绑定队列和交换机
// 需求： 所有err 级别的日志存入数据库，所有order 系统的日志存入到数据库
channel.queueBind(TOPIC_QUEUE_1,TOPIC_EXCHAGE,"#.error");
channel.queueBind(TOPIC_QUEUE_1,TOPIC_EXCHAGE,"order.*");
// QUEUE2 不管什么级别的信息，都打印到控制台
channel.queueBind(TOPIC_QUEUE_2,TOPIC_EXCHAGE,"*.*");
// 发送信息 routingKey 是 系统的名称.日志的级别
String message = "日志信息: 张三调用订单的findAll() 方法，系统级别 info " ;
channel.basicPublish(TOPIC_EXCHAGE, "order.info", null, message.getBytes());
System.out.println("已发送消息：" + message);
// 发送信息
message = "日志信息: 张三调用库存的get() 方法，系统级别 error" ;
channel.basicPublish(TOPIC_EXCHAGE, "kucun.error", null, message.getBytes());
System.out.println("已发送消息：" + message);
// 发送信息
message = "日志信息: 张三调用库存的findAll() 方法，系统级别 info" ;
channel.basicPublish(TOPIC_EXCHAGE, "kucun.info", null, message.getBytes());
System.out.println("已发送消息：" + message);
// 关闭资源
channel.close();
connection.close();
}
}
两个消费者：
consumer1 接受所有日志级别为error的消息和所有order 系统的消息。
import com.boshrong.rabbitmq.ConnectionUtil;
import com.rabbitmq.client.*;
import java.io.IOException;
public class Consumer1 {
public static void main(String[] args) throws Exception {
Connection connection = ConnectionUtil.getConnection();
// 创建频道
Channel channel = connection.createChannel();
//声明交换机
channel.exchangeDeclare(producer_topic.TOPIC_EXCHAGE, BuiltinExchangeType.TOPIC);
// 声明（创建）队列
channel.queueDeclare(producer_topic.TOPIC_QUEUE_1, true, false, false, null);
//队列绑定交换机
// consumer1 接受所有日志级别为error的消息和所有order 系统的消息。
channel.queueBind(producer_topic.TOPIC_QUEUE_1, producer_topic.TOPIC_EXCHAGE, "#.error");
channel.queueBind(producer_topic.TOPIC_QUEUE_1, producer_topic.TOPIC_EXCHAGE, "order.*");
//创建消费者；并设置消息处理
DefaultConsumer consumer = 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"));
}
};
//监听消息
channel.basicConsume(producer_topic.TOPIC_QUEUE_1, true, consumer);
}
}
Consumer2 ：Consumer2 接受所有类型的消息
import com.boshrong.rabbitmq.ConnectionUtil;
import com.rabbitmq.client.*;
import java.io.IOException;
public class Consumer2 {
public static void main(String[] args) throws Exception {
Connection connection = ConnectionUtil.getConnection();
// 创建频道
Channel channel = connection.createChannel();
//声明交换机
channel.exchangeDeclare(producer_topic.TOPIC_EXCHAGE, BuiltinExchangeType.TOPIC);
// 声明（创建）队列
channel.queueDeclare(producer_topic.TOPIC_QUEUE_2, true, false, false, null);
//队列绑定交换机
//Consumer2 接受所有类型的消息
channel.queueBind(producer_topic.TOPIC_QUEUE_2, producer_topic.TOPIC_EXCHAGE, "*.*");
//创建消费者；并设置消息处理
DefaultConsumer consumer = 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("消费者2-接收到的消息为：" + new String(body, "utf-8"));
}
};
//监听消息
channel.basicConsume(producer_topic.TOPIC_QUEUE_2, true, consumer);
}
}
consume1运行结果
consume2 运行结果
3.6 模式总结RabbitMQ工作模式：
1、简单模式 HelloWorld
一个生产者、一个消费者，不需要设置交换机（使用默认的交换机）
2、工作队列模式 Work Queue
一个生产者、多个消费者（竞争关系），不需要设置交换机（使用默认的交换机）
3、发布订阅模式 Publish/subscribe
需要设置类型为fanout的交换机，并且交换机和队列进行绑定，当发送消息到交换机后，交换机会将消息发送到绑定的队列
4、路由模式 Routing
需要设置类型为direct的交换机，交换机和队列进行绑定，并且指定routing key，当发送消息到交换机后，交换机会根据routing key将消息发送到对应的队列
5、通配符模式 Topic
需要设置类型为topic的交换机，交换机和队列进行绑定，并且指定通配符方式的routing key，当发送消息到交换机后，交换机会根据routing key将消息发送到对应的队列
欢迎分享，转载请注明来源：内存溢出
微信扫一扫
支付宝扫一扫
评论列表（0条）