RabbitMQ 工作模式二

之前写了WORKQUEUES 跟 Publish/Subscribe 俩种模式 ,RabbitMQ 工作模式一

Routing 工作模式

特点

每个消费者监听自己的队列，并且设置routingkey

生产者将消息发给交换机，由交换机根据routingkey来转发消息到指定的队列

说白了,就是在publish/subscribe 工作模式的基础上加一层筛选,判断

 if(队列.routingkey == 生产者.routingkey)    发送消息给duilie

下面是我写的生产者

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;public class Producer03 {//队列名称private static final String QUEUE_INFORM_TEST1 = "queue_inform_test1";private static final String QUEUE_INFORM_TEST2 = "queue_inform_test2";public final static String  EXCHANGE_ROUTING_INFORM = "exchange_routing_inform";public static void main(String[] args) throws Exception{ConnectionFactory factory = new ConnectionFactory();//设置ipfactory.setHost("localhost");//设置端口factory.setPort(5672);//设置账号密码factory.setUsername("guest");  //默认账号密码都是guestfactory.setPassword("guest");//设置虚拟空间factory.setVirtualHost("/");//虚拟机默认的虚拟名称为/ , 虚拟机相当于一个独立的服务器Connection connection = factory.newConnection();Channel channel = connection.createChannel();//创建一个通道/*** 定义交换机*  param1 :  交换机名称*  param2 :   交换机类型**/channel.exchangeDeclare(EXCHANGE_ROUTING_INFORM, BuiltinExchangeType.DIRECT);/***    定义消息队列**    String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments**/channel.queueDeclare(QUEUE_INFORM_TEST1,true ,false ,false ,null);channel.queueDeclare(QUEUE_INFORM_TEST2,true ,false ,false ,null);//绑定交换机跟队列/*String queue, String exchange, String routingKey*/channel.queueBind(QUEUE_INFORM_TEST1,EXCHANGE_ROUTING_INFORM ,QUEUE_INFORM_TEST1);channel.queueBind(QUEUE_INFORM_TEST2,EXCHANGE_ROUTING_INFORM ,QUEUE_INFORM_TEST2);String message = "";//给队列发送消息for (int i = 0; i < 9; i++) {//
//            String exchange, String routingKey, BasicProperties props, byte[] bodymessage = "人间有百媚千红，唯你是我情之所钟。第"+i+"条消息";//给test1 发送9条消息channel.basicPublish(EXCHANGE_ROUTING_INFORM,QUEUE_INFORM_TEST1 ,null ,message.getBytes() );}for (int i = 0; i < 5; i++) {//
//            String exchange, String routingKey, BasicProperties props, byte[] bodymessage = "你别回头看我了，走吧，山高水长，可别再碰到我这么喜欢你的人了。第"+i+"条消息";//给test2 发送5条消息channel.basicPublish(EXCHANGE_ROUTING_INFORM,QUEUE_INFORM_TEST2 ,null ,message.getBytes() );}channel.close();connection.close();}}

设置了交换机名称,及路由的交换机类型

channel.exchangeDeclare(EXCHANGE_ROUTING_INFORM, BuiltinExchangeType.DIRECT);

--------------------------------------------------------------------------------------------------------------------------------------

声明了俩个队列...分别为 test1, test2

*** String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments** param1: 队列名称* param2: 是否持久化* param3 : 是否独占此队列* param4 : 队列不用是否自动删除*  param5 : 参数*/

channel.queueDeclare(QUEUE_INFORM_Test1,true ,false ,false ,null );channel.queueDeclare(QUEUE_INFORM_Test2,true ,false ,false ,null );

------------------------------------------------------------------------------------------------------------

队列和交换机绑定

//交换机和队列绑定
/*** String queue, String exchange, String routingKey* param1 :  队列名称* exchange :  交换机* routingKey : 给队列添加一个 路由key,交换机发送消息时根据填写的路由key 来判断,如果填写的key 跟 队列的路由key 相同,那么就会发送消息给此队列 */

  channel.queueBind(QUEUE_INFORM_TEST1,EXCHANGE_ROUTING_INFORM ,QUEUE_INFORM_TEST1);channel.queueBind(QUEUE_INFORM_TEST2,EXCHANGE_ROUTING_INFORM ,QUEUE_INFORM_TEST2);

-------------------------------------------------------------------------

发送消息

/*** String exchange, String routingKey, BasicProperties props, byte[] body**   param1  交换机名称*   param2  根据key名称将消息转发到具体的队列，这里填写队列名称表示消息将发到此队列*   param3  参数*   param4 传递的字符串***/

 channel.basicPublish(EXCHANGE_FANOUT_INFORM,"" , null, message.getBytes());

``````````````````````````````````````````````````````````````````````````````````````````````````````````````

以上是生产者的代码 ,下面提供消费者的代码

import com.rabbitmq.client.*;import java.io.IOException;public class ConsumerTest1 {//队列名称private static final String QUEUE_INFORM_TEST1 = "queue_inform_test1";public static void main(String[] args){Connection connection = null;Channel channel = null;try {ConnectionFactory factory = new ConnectionFactory();//设置ipfactory.setHost("localhost");//设置端口factory.setPort(5672);//设置账号密码factory.setUsername("guest");  //默认账号密码都是guestfactory.setPassword("guest");//设置虚拟空间factory.setVirtualHost("/");//虚拟机默认的虚拟名称为/ , 虚拟机相当于一个独立的服务器//创建与RabbitMQ服务的TCP连接connection = factory.newConnection();//创建连接通道 ,每个连接可以创建多个通道,每个通道只有一个会话channel = connection.createChannel();//            创建默认消费方法DefaultConsumer consumer = new DefaultConsumer(channel) {//                重写监听方法@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("receive message.."+new String(body,"utf-8"));}};//            监听队列/*** String queue, boolean autoAck, Consumer callback** param1 :  队列名称* param2 :   是否自动回复,接收到消息会自动恢复mq收到了,mq会删除消息,如果拒绝的话需要手动回复,不回复的话会导致mq不删除被消费过的消息,一直存在* param3 : 消费对象,,包含消费方法**/channel.basicConsume(QUEUE_INFORM_TEST1,true,consumer);}catch (Exception e){}}
}

消费者二

import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DefaultConsumer;import com.rabbitmq.client.*;import java.io.IOException;public class ConsumerTest2 {private static final String QUEUE_INFORM_TEST2 = "queue_inform_test2";public static void main(String[] args){Connection connection = null;Channel channel = null;try {ConnectionFactory factory = new ConnectionFactory();//设置ipfactory.setHost("localhost");//设置端口factory.setPort(5672);//设置账号密码factory.setUsername("guest");  //默认账号密码都是guestfactory.setPassword("guest");//设置虚拟空间factory.setVirtualHost("/");//虚拟机默认的虚拟名称为/ , 虚拟机相当于一个独立的服务器//创建与RabbitMQ服务的TCP连接connection = factory.newConnection();//创建连接通道 ,每个连接可以创建多个通道,每个通道只有一个会话channel = connection.createChannel();//            创建默认消费方法DefaultConsumer consumer = new DefaultConsumer(channel) {//                重写监听方法@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("receive message.."+new String(body,"utf-8"));}};//            监听队列/*** String queue, boolean autoAck, Consumer callback** param1 :  队列名称* param2 :   是否自动回复,接收到消息会自动恢复mq收到了,mq会删除消息,如果拒绝的话需要手动回复,不回复的话会导致mq不删除被消费过的消息,一直存在* param3 : 消费对象,,包含消费方法**/channel.basicConsume(QUEUE_INFORM_TEST2,true , consumer);}catch (Exception e){}}
}

消费者的代码其实很简单,就是监听而已.只需要指定一下监听的队列就行并提供一个执行的方法 ,就是下面这句

channel.basicConsume(QUEUE_INFORM_TEST2,true , consumer);

测试

先启动producer,否则会报错,因为consumer 中没有声明队列,并且没有中rabbitmq中发现队列,就会抛出异常

生产者会给 test1 发送 9条消息给test 发送 5条消息

然后我们依次启动俩个消费者

然后看处理结果

Topics 工作模式

通配符工作模式

每个消费者监听自己的队列，并且设置带统配符的routingkey

生产者将消息发给broker，由交换机根据routingkey来转发消息到指定的队列。

统配符规则：

符号#可以匹配多个词，

符号*可以匹配一个词语。

因为通配符感觉比较难讲,所以我在网上找了一个充值的案例

场景

用户充值完成, email 的用户接收email的提示, sms的用户接收sms的提示.. 设置两种通知类型都接收的则两种通知都有效。

大致思路

设置路由匹配规则

邮件的匹配规则 "inform.#.email.#" sms的匹配规则 "inform.#.sms.#"

给只接收email的用户发送消息路由设置 inform.email

给只接收 sm的用户发送消息路由设置 inform.sms

给都接收的用户发送消息 inform.sms.email 不懂的话看一下匹配规则,,有点取巧来着

下面是我写的生产者

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;public class Producer03 {//队列名称private static final String QUEUE_INFORM_EMAIL = "queue_inform_email";private static final String QUEUE_INFORM_SMS = "queue_inform_sms";public final static String  EXCHANGE_ROUTING_INFORM = "exchange_topic_inform";public static void main(String[] args) throws Exception{ConnectionFactory factory = new ConnectionFactory();//设置ipfactory.setHost("localhost");//设置端口factory.setPort(5672);//设置账号密码factory.setUsername("guest");  //默认账号密码都是guestfactory.setPassword("guest");//设置虚拟空间factory.setVirtualHost("/");//虚拟机默认的虚拟名称为/ , 虚拟机相当于一个独立的服务器Connection connection = factory.newConnection();Channel channel = connection.createChannel();//创建一个通道/*** 定义交换机*  param1 :  交换机名称*  param2 :   交换机类型**/channel.exchangeDeclare(EXCHANGE_ROUTING_INFORM, BuiltinExchangeType.TOPIC);/***    定义消息队列**    String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments**/channel.queueDeclare(QUEUE_INFORM_EMAIL,true ,false ,false ,null);channel.queueDeclare(QUEUE_INFORM_SMS,true ,false ,false ,null);//绑定交换机跟队列/*String queue, String exchange, String routingKey*/channel.queueBind(QUEUE_INFORM_EMAIL,EXCHANGE_ROUTING_INFORM ,"inform.#.email.#");channel.queueBind(QUEUE_INFORM_SMS,EXCHANGE_ROUTING_INFORM ,"inform.#.sms.#");String message = "";//给接收邮件的人发邮件for (int i = 0; i < 9; i++) {//
//            String exchange, String routingKey, BasicProperties props, byte[] bodymessage = "记忆最让人崩溃的地方，也许就在于它的猝不及防在某个祥和的午后，你正吃着火锅唱着歌，那些尖利的记忆碎片就像潮水突然涌进你到脑海里，让你闪躲不及。";//给test1 发送9条消息channel.basicPublish(EXCHANGE_ROUTING_INFORM,"inform.email" ,null ,message.getBytes() );}//给接收短信的人发短信for (int i = 0; i < 5; i++) {//
//            String exchange, String routingKey, BasicProperties props, byte[] bodymessage = "发送了一百条短信,九十九条都是你";//给test2 发送5条消息channel.basicPublish(EXCHANGE_ROUTING_INFORM,"inform.sms" ,null ,message.getBytes() );}//        给都接收的人发送 for (int i = 0; i < 3; i++) {message = "我回你是秒回,你回我是轮回";//给test2 发送5条消息channel.basicPublish(EXCHANGE_ROUTING_INFORM,"inform.sms.email" ,null ,message.getBytes() );}channel.close();connection.close();}}

-----------------------

定义交换机,并设置交换机的类型为通配符模式

/*** 定义交换机*  param1 :  交换机名称*  param2 :   交换机类型**/

channel.exchangeDeclare(EXCHANGE_ROUTING_INFORM, BuiltinExchangeType.TOPIC);

---------------------------------------------------

绑定交换机跟队列,并配置队列的路由通配符规则

  channel.queueBind(QUEUE_INFORM_EMAIL,EXCHANGE_ROUTING_INFORM ,"inform.#.email.#");channel.queueBind(QUEUE_INFORM_SMS,EXCHANGE_ROUTING_INFORM ,"inform.#.sms.#");

--------------------------------------------------------------------------------------------

发送消息给broker 并发送路由key

/*** 参数明细

* 1、交换机名称，不指令使用默认交换机名称 Default Exchang

* 2、routingKey（路由key），根据key名称将消息转发到具体的队列，这里填写队列名称表示消息将发到此队列

* 3、消息属性

* 4、消息内容*/

 channel.basicPublish(EXCHANGE_ROUTING_INFORM,"inform.email" ,null ,message.getBytes() );channel.basicPublish(EXCHANGE_ROUTING_INFORM,"inform.sms" ,null ,message.getBytes() );
channel.basicPublish(EXCHANGE_ROUTING_INFORM,"inform.sms.email" ,null ,message.getBytes());

消息监听者1

import com.rabbitmq.client.*;import java.io.IOException;public class ConsumerTest1 {//队列名称private static final String QUEUE_INFORM_EMAIL = "queue_inform_email";public static void main(String[] args){Connection connection = null;Channel channel = null;try {ConnectionFactory factory = new ConnectionFactory();//设置ipfactory.setHost("localhost");//设置端口factory.setPort(5672);//设置账号密码factory.setUsername("guest");  //默认账号密码都是guestfactory.setPassword("guest");//设置虚拟空间factory.setVirtualHost("/");//虚拟机默认的虚拟名称为/ , 虚拟机相当于一个独立的服务器//创建与RabbitMQ服务的TCP连接connection = factory.newConnection();//创建连接通道 ,每个连接可以创建多个通道,每个通道只有一个会话channel = connection.createChannel();//            创建默认消费方法DefaultConsumer consumer = new DefaultConsumer(channel) {//                重写监听方法@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("receive message.."+new String(body,"utf-8"));}};//            监听队列/*** String queue, boolean autoAck, Consumer callback** param1 :  队列名称* param2 :   是否自动回复,接收到消息会自动恢复mq收到了,mq会删除消息,如果拒绝的话需要手动回复,不回复的话会导致mq不删除被消费过的消息,一直存在* param3 : 消费对象,,包含消费方法**/channel.basicConsume(QUEUE_INFORM_EMAIL,true,consumer);}catch (Exception e){}}
}

消息监听二

import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DefaultConsumer;import com.rabbitmq.client.*;import java.io.IOException;public class ConsumerTest2 {private static final String QUEUE_INFORM_SMS= "queue_inform_sms";public static void main(String[] args){Connection connection = null;Channel channel = null;try {ConnectionFactory factory = new ConnectionFactory();//设置ipfactory.setHost("localhost");//设置端口factory.setPort(5672);//设置账号密码factory.setUsername("guest");  //默认账号密码都是guestfactory.setPassword("guest");//设置虚拟空间factory.setVirtualHost("/");//虚拟机默认的虚拟名称为/ , 虚拟机相当于一个独立的服务器//创建与RabbitMQ服务的TCP连接connection = factory.newConnection();//创建连接通道 ,每个连接可以创建多个通道,每个通道只有一个会话channel = connection.createChannel();//            创建默认消费方法DefaultConsumer consumer = new DefaultConsumer(channel) {//                重写监听方法@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("receive message.."+new String(body,"utf-8"));}};//            监听队列/*** String queue, boolean autoAck, Consumer callback** param1 :  队列名称* param2 :   是否自动回复,接收到消息会自动恢复mq收到了,mq会删除消息,如果拒绝的话需要手动回复,不回复的话会导致mq不删除被消费过的消息,一直存在* param3 : 消费对象,,包含消费方法**/channel.basicConsume(QUEUE_INFORM_SMS,true , consumer);}catch (Exception e){}}
}

其实监听者代码都不要变来着,都是通用的

测试

先启动生产者,后启动监听者.....理由是如果先启动监听者,队列没创建,就会报错

测试结果

sms的打印台打印结果

email 打印台打印结果

其实搞懂了路由模式后这个就很容易理解这个....

Header模式

header 模式其实跟路由模式很像,他们不同的是header模式取消routingkey，使用header中的 key/value（键值对）匹配队列

了解了很多不同的模式,其实你会发现,代码很多都是相同的....这里我就不粘贴全部代码,就只给不同的代码了

绑定交换机跟队列的代码

  HashMap<String, Object> header_email = new HashMap<>();header_email.put("inform_type", "cms");HashMap<String, Object> header_sms = new HashMap<>();header_sms.put("inform_type", "sms");channel.queueBind(QUEUE_INFORM_EMAIL,EXCHANGE_ROUTING_INFORM ,"",header_email);channel.queueBind(QUEUE_INFORM_SMS,EXCHANGE_ROUTING_INFORM ,"",header_sms);

发送消息的代码

message = "记忆最让人崩溃的地方，也许就在于它的猝不及防在某个祥和的午后，你正吃着火锅唱着歌，那些尖利的记忆碎片就像潮水突然涌进你到脑海里，让你闪躲不及。";HashMap<String, Object> header = new HashMap<>();header.put("inform_type", "sms");  //匹配sms通知消费者绑定的headerAMQP.BasicProperties.Builder properties = new AMQP.BasicProperties.Builder();properties.headers(header)//给test1 发送9条消息channel.basicPublish(EXCHANGE_ROUTING_INFORM,"" ,properties.build() ,message.getBytes() );

消费者都是一样的,改一下队列名称就行