【RabbitMQ】RabbbitMQ的六种工作模式以及代码实现

悠悠 2023-09-27 17:25 104阅读 0赞

**目录**

一、交换机类型

二、简单模式

1、介绍

2、代码实现

三、Work Queues工作队列模式

1、介绍

2、代码实现

四、Pub/Sub订阅模式

1、介绍

2、代码实现

五、Routing路由模式

1、介绍

2、代码实现

六、Topics通配符模式

1、介绍

2、代码实现

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

## 一、交换机类型 ##

在之前的文章里我们了解了什么是交换机，当有生产者把消息发给MQ服务器时，消息会通过交换机分发给不同的消息队列里，那么交换机是如何分发消息的，我们就需要了解交换机的几种类型

<table style="width:500px;"> 
 <tbody> 
  <tr> 
   <td>类型</td> 
   <td>说明</td> 
  </tr> 
  <tr> 
   <td>fanout</td> 
   <td>广播类型，交换机会把收到的消息分发给每一与他绑定的队列</td> 
  </tr> 
  <tr> 
   <td>direct</td> 
   <td>定向类型，交换机会根据路由将消息分发给与它绑定的指定的匹配队列</td> 
  </tr> 
  <tr> 
   <td>topic</td> 
   <td>通配符类型，交换机会根据通配符匹配后分发给与他绑定的指定的匹配队列</td> 
  </tr> 
 </tbody> 
</table>

## 二、简单模式 ##

### 1、介绍 ###

RabbitMQ的简单模式是一对一即，一个生产者生产消息后不经交换机直接给指定的队列供消费者消费![cc4ad89f73e34a52a9e3a0d42fd61148.png][]

### 2、代码实现 ###

他的不需要交换机进行分发，所以我们代码实现整体过程为

生产者：建立连接-》创建队列-》构造发送消息-》关闭连接

import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    public class Producer {
        private static final String QUEUE = "easyQueue";
    
        public static void main(String[] args) throws IOException, TimeoutException {
            // 1.创建连接工厂 配置连接工厂
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("127.0.0.1");
            factory.setPort(5672);
            factory.setVirtualHost("/DemoVirtualHost");
            factory.setUsername("guest");
            factory.setPassword("guest");
    
            // 2.建立连接 获取连接通道
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
    
            // 3.创建队列
            channel.queueDeclare(QUEUE,true,false,false,null);
    
            // 4.发送消息
            String body = "hello mq";
            channel.basicPublish("",QUEUE,null,body.getBytes());
    
            // 5.关闭连接
            channel.close();
            connection.close();
        }
    }

运行代码即可发送消息到服务器![720ffc50619a4c53b2a9cf61cc07f77f.png][]

消费者：建立连接监听队列获取消息

import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    public class Consumer {
        private static final String QUEUE = "easyQueue";
    
        public static void main(String[] args) throws IOException, TimeoutException {
            // 1.创建连接工厂 配置连接工厂
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("127.0.0.1");
            factory.setPort(5672);
            factory.setVirtualHost("/DemoVirtualHost");
            factory.setUsername("guest");
            factory.setPassword("guest");
    
            // 2.建立连接
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
    
            // 3.监听队列获取消息
            com.rabbitmq.client.Consumer consumer = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    System.out.println("消费者消费消息:" + new String(body));
                }
            };
            channel.basicConsume(QUEUE,true,consumer);
        }
    }

运行代码![b09ffeba0bf54375a0cbe0425d13af49.png][]

完成消费，这就是RabbitMQ工作模式的简单模式，他不需要交换机来去对消息进行分发

## 三、Work Queues工作队列模式 ##

### 1、介绍 ###

工作队列模式相比简单模式，他的处理任务速度在一定情况下会更快，因为他相比简单模式单一消费者而言它增加了消费者个数

![ebfb22c53fbd496b9bc9e356e621ff0f.png][]

### 2、代码实现 ###

它同样也不需要交换机对收到的消息进行分发，代码实现与简单模式基本相同，唯一不同点是他相比简单模式多了一个消费者

生产者

import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    public class Producer {
        private static final String QUEUE = "workQueue";
    
        public static void main(String[] args) throws IOException, TimeoutException {
            // 1.创建连接工厂 配置连接工厂
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("127.0.0.1");
            factory.setPort(5672);
            factory.setVirtualHost("/DemoVirtualHost");
            factory.setUsername("guest");
            factory.setPassword("guest");
    
            // 2.建立连接 获取连接通道
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
    
            // 3.创建队列
            channel.queueDeclare(QUEUE,true,false,false,null);
    
            // 4.发送消息
            String body = "hello mq";
            channel.basicPublish("",QUEUE,null,body.getBytes());
    
            // 5.关闭连接
            channel.close();
            connection.close();
        }
    }

消费者1

import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    public class Consumer {
        private static final String QUEUE = "workQueue";
    
        public static void main(String[] args) throws IOException, TimeoutException {
            // 1.创建连接工厂 配置连接工厂
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("127.0.0.1");
            factory.setPort(5672);
            factory.setVirtualHost("/DemoVirtualHost");
            factory.setUsername("guest");
            factory.setPassword("guest");
    
            // 2.建立连接
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
    
            // 3.监听队列获取消息
            com.rabbitmq.client.Consumer consumer = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    System.out.println("消费者消费消息:" + new String(body));
                }
            };
            channel.basicConsume(QUEUE,true,consumer);
        }
    }

消费者2：

## 四、Pub/Sub订阅模式 ##

### 1、介绍 ###

他相比之前两种模式引入了交换机，当MQ服务器收到消息后，交换机会把收到的消息分发给每一个绑定了该交换机的队列给不同的消费者进行消费，之前的文章里我们是通过控制台进行队列与交换机的绑定，这里我们会使用Java代码来进行绑定![14c2002718e5424ea0acaf503bf48064.png][]

比如当用户下单后我们需要去给用户通过邮箱以及短信发送回购优惠券信息时，我们就可以通过该模式，分别创建邮箱队列以及短信队列，然后生产者将用户信息发给MQ服务器经过交换机进入邮箱与短信队列让不同的消费者去消费通知。

### 2、代码实现 ###

与前两种模式不同的时，我们在实现该模式的生产者时建立连接后需要去创建交换机，由于交换机不同的类型其工作模式不同，所以我们创建交换机时 需要去指定他的类型，创建完成后我们还需要去将邮箱队列与短信队列与该交换机进行绑定

创建交换机需要使用该方法exchangeDeclare

<table style="width:500px;"> 
 <tbody> 
  <tr> 
   <td>参数</td> 
   <td>说明</td> 
  </tr> 
  <tr> 
   <td>exchange</td> 
   <td>交换机名</td> 
  </tr> 
  <tr> 
   <td>type</td> 
   <td>交换机类型</td> 
  </tr> 
  <tr> 
   <td>durable</td> 
   <td>是否持久化到内存，一般为true</td> 
  </tr> 
  <tr> 
   <td>autoDelete</td> 
   <td>是否自动删除，一般为false</td> 
  </tr> 
  <tr> 
   <td>internal</td> 
   <td>内部使用参数，一般为false</td> 
  </tr> 
  <tr> 
   <td>arguments</td> 
   <td>参数信息，一般为null</td> 
  </tr> 
 </tbody> 
</table>

队列绑定交换机需要用到这个queueBind（队列名，交换机名，路由名--如果交换机类型为fanout则为空字符串）方法

生产者：

import com.rabbitmq.client.BuiltinExchangeType;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    public class Producer {
        private static final String EXCHANGE_NAME = "pubSubExchange";
        private static final String EMAIL_QUEUE = "email";
        private static final String SMS_QUEUE = "sms";
    
        public static void main(String[] args) throws IOException, TimeoutException {
            // 1.创建连接工厂 配置连接工厂
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("127.0.0.1");
            factory.setPort(5672);
            factory.setVirtualHost("/DemoVirtualHost");
            factory.setUsername("guest");
            factory.setPassword("guest");
    
            // 2.建立连接
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
    
            // 3.创建交换机
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT,true,false,false,null);
    
            // 4.创建队列
            channel.queueDeclare(EMAIL_QUEUE,true,false,false,null);
            channel.queueDeclare(SMS_QUEUE,true,false,false,null);
    
            // 5.绑定交换机
            channel.queueBind(EMAIL_QUEUE,EXCHANGE_NAME,"");
            channel.queueBind(SMS_QUEUE,EXCHANGE_NAME,"");
    
            // 6.发送信息
            channel.basicPublish(EXCHANGE_NAME,"",null,"hello".getBytes());
    
            // 7.关闭连接
            channel.close();
            connection.close();
        }
    }

运行查看服务器![33fa509d937d4639b0500e40480e27f5.png][]

此时两个消费者只需修改监听的队列即可

email消费者：

import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    public class EmailConsumer {
        private static final String EMAIL_QUEUE = "email";
        private static final String SMS_QUEUE = "sms";
    
        public static void main(String[] args) throws IOException, TimeoutException {
            // 1.创建连接工厂 配置工厂
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("127.0.0.1");
            factory.setPort(5672);
            factory.setVirtualHost("/DemoVirtualHost");
            factory.setUsername("guest");
            factory.setPassword("guest");
    
            // 2.建立连接
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
    
            // 3.监听队列获取消息
            Consumer consumer = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    System.out.println("邮件已发送" + new String(body));
                }
            };
            channel.basicConsume(EMAIL_QUEUE,true,consumer);
        }
    }

运行代码![e555223c723248c38b64a1a62d91b6c7.png][]

sms消费者：

import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    public class EmailConsumer {
        private static final String EMAIL_QUEUE = "email";
        private static final String SMS_QUEUE = "sms";
    
        public static void main(String[] args) throws IOException, TimeoutException {
            // 1.创建连接工厂 配置工厂
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("127.0.0.1");
            factory.setPort(5672);
            factory.setVirtualHost("/DemoVirtualHost");
            factory.setUsername("guest");
            factory.setPassword("guest");
    
            // 2.建立连接
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
    
            // 3.监听队列获取消息
            Consumer consumer = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    System.out.println("已发送" + new String(body));
                }
            };
            channel.basicConsume(SMS_QUEUE,true,consumer);
        }
    }

## 五、Routing路由模式 ##

### 1、介绍 ###

上述用户下单发送回购优惠券需求时，如果有的用户他注册时没有设置邮箱或者电话，我们不需要去给没有的邮箱发送，此时使用订阅模式是不可取的，使用路由模式是可行的，那么为什么路由模式可以呢，让我们来看。路由模式与订阅模式相同的是都需要使用交换机，不同的是当收到生产者的消息时，订阅模式会分发给每一个与之绑定的队列，而路由模式则不同，他会按照消息指定的路由去匹配到相应的队列从而发送给指定的队列

![b1de0aac0984445e9258ca1519142456.png][]

### 2、代码实现 ###

由上述路由模式图，我们在发送消息前，需要先创建一个交换机且路由模式交换机类型不在是fanout而是direct路由模式，还需要去创建队列，然后绑定交换机，在绑定交换机的时候我们需要去给队列设置对应的路由，且在发送消息时指定路由

生产者：

import com.rabbitmq.client.BuiltinExchangeType;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    public class Producer {
        private static final String EXCHANGE_NAME = "rout-ex";
        private static final String EMAIL_QUEUE = "email";
        private static final String SMS_EMAIL = "sms";
    
        public static void main(String[] args) throws IOException, TimeoutException {
            // 1.创建连接工厂 配置工厂
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("127.0.0.1");
            factory.setPort(5672);
            factory.setVirtualHost("/DemoVirtualHost");
            factory.setUsername("guest");
            factory.setPassword("guest");
    
            // 2.建立连接
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
    
            // 3.创建交换机
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT,true,false,false,null);
    
            // 4.创建队列
            channel.queueDeclare(EMAIL_QUEUE,true,false,false,null);
            channel.queueDeclare(SMS_EMAIL,true,false,false,null);
    
            // 5.绑定交换机
            channel.queueBind(EMAIL_QUEUE,EXCHANGE_NAME,"email");
            channel.queueBind(SMS_EMAIL,EXCHANGE_NAME,"sms");
    
            // 6.发送消息
            channel.basicPublish(EXCHANGE_NAME,"email",null,"routing".getBytes());
    
            // 7.关闭连接
            channel.close();
            connection.close();
        }
    }

发送消息指定了路由为email，此时我们运行代码查看服务器![ec38ba035b5e40cc9dd033ecc55b8b5d.png][]

消费者：

修改监听队列名即可

import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    public class Consumer {
        public static void main(String[] args) throws IOException, TimeoutException {
            // 1.获取连接工厂
            ConnectionFactory connectionFactory = new ConnectionFactory();
    
            // 2.设置参数：队列、交换机、虚拟机、端口、IP
            connectionFactory.setHost("127.0.0.1");
            connectionFactory.setPort(5672);
            connectionFactory.setVirtualHost("/DemoVirtualHost");
            connectionFactory.setUsername("1886");
            connectionFactory.setPassword("1886");
    
            // 3.建立连接 Connection ---- channel
            Connection connection = connectionFactory.newConnection();
            Channel channel = connection.createChannel();
    
            // 4.获取消息
            com.rabbitmq.client.Consumer consumer = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    System.out.println("email消费");
                }
            };
            channel.basicConsume("email",true,consumer);
    
        }
    }

![dd11e38fa3704fa2ae178471ae310e13.png][]

## 六、Topics通配符模式 ##

### 1、介绍 ###

上述路由模式我们知道了消息可以指定路由进而到达指定的队列，但路由只有一个所以该消息只能进入一个队列，而后续如果有一个消息需要进入几个队列的需求时订阅模式与路由模式就不能完成，此时我们可以使用通配符模式进行匹配，而该模式的交换机类型为通配符类型的topic。

通配符有两种

<table style="width:500px;"> 
 <tbody> 
  <tr> 
   <td>种类</td> 
   <td>介绍</td> 
  </tr> 
  <tr> 
   <td>*（system.*）</td> 
   <td>*指的是匹配一个单词，（system.*）是指路由前缀为system.且后面有一个单词即可</td> 
  </tr> 
  <tr> 
   <td>#（#.error）</td> 
   <td>#指的是匹配0个或多个单词，（#.error）是指路由前面没有或有若干单词，后面为.error</td> 
  </tr> 
 </tbody> 
</table>

如果此时服务器有两个队列一个为存放所有分布式系统的error日志，一个只用于存放A系统的所有等级日志 ，此时前者路由配置为\#.error，后者可为A.\*

### 2、代码实现 ###

生产者：

我们实现上述业务时可对队列进行上述路由配置，此时我们将消息路由设置为B系统下的error日志：B.error，该消息就会进入error队列

import com.rabbitmq.client.BuiltinExchangeType;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    public class Producer {
        private static final String EXCHANGE_NAME = "topic-ex";
        private static final String ERROR_QUEUE = "error";
        private static final String SYSTEM_LOG = "a-system-logging";
    
        public static void main(String[] args) throws IOException, TimeoutException {
            // 1.创建连接工厂 配置工厂
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("127.0.0.1");
            factory.setPort(5672);
            factory.setVirtualHost("/DemoVirtualHost");
            factory.setUsername("guest");
            factory.setPassword("guest");
    
            // 2.建立连接
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
    
            // 3.创建交换机
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC,true,false,false,null);
    
            // 4.创建队列
            channel.queueDeclare(ERROR_QUEUE,true,false,false,null);
            channel.queueDeclare(SYSTEM_LOG,true,false,false,null);
    
            // 5.绑定交换机
            channel.queueBind(ERROR_QUEUE,EXCHANGE_NAME,"#.error");  // 所有系统的error日志
            channel.queueBind(SYSTEM_LOG,EXCHANGE_NAME,"A.*");       // A系统的所有日志
    
            // 6.发送消息
            channel.basicPublish(EXCHANGE_NAME,"B.error",null,"b系统的error日志".getBytes());
    
            //  7.关闭连接
            channel.close();
            connection.close();
        }
    }

![a76fa2e02cbe492ea428f41737450b0f.png][]

消费者：

消费者只需要修改监听的队列名即可

import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    public class Consumer {
        public static void main(String[] args) throws IOException, TimeoutException {
            // 1.创建连接工厂 配置工厂
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("127.0.0.1");
            factory.setPort(5672);
            factory.setVirtualHost("/DemoVirtualHost");
            factory.setUsername("guest");
            factory.setPassword("guest");
    
            // 2.建立连接
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
    
            // 3.监听队列获取消息
            com.rabbitmq.client.Consumer consumer = new DefaultConsumer(channel){
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    System.out.println("处理日志" + new String(body));
                }
            };
            channel.basicConsume("error",true,consumer);
    
        }
    }

运行代码即可查看![c3cee6c1fcb84fb69ff345ad8a5287db.png][]

![fb69ed53c3234273b86bcba2d9a74071.png][]

[cc4ad89f73e34a52a9e3a0d42fd61148.png]: https://img-blog.csdnimg.cn/cc4ad89f73e34a52a9e3a0d42fd61148.png
[720ffc50619a4c53b2a9cf61cc07f77f.png]: https://img-blog.csdnimg.cn/720ffc50619a4c53b2a9cf61cc07f77f.png
[b09ffeba0bf54375a0cbe0425d13af49.png]: https://img-blog.csdnimg.cn/b09ffeba0bf54375a0cbe0425d13af49.png
[ebfb22c53fbd496b9bc9e356e621ff0f.png]: https://img-blog.csdnimg.cn/ebfb22c53fbd496b9bc9e356e621ff0f.png
[14c2002718e5424ea0acaf503bf48064.png]: https://img-blog.csdnimg.cn/14c2002718e5424ea0acaf503bf48064.png
[33fa509d937d4639b0500e40480e27f5.png]: https://img-blog.csdnimg.cn/33fa509d937d4639b0500e40480e27f5.png
[e555223c723248c38b64a1a62d91b6c7.png]: https://img-blog.csdnimg.cn/e555223c723248c38b64a1a62d91b6c7.png
[b1de0aac0984445e9258ca1519142456.png]: https://img-blog.csdnimg.cn/b1de0aac0984445e9258ca1519142456.png
[ec38ba035b5e40cc9dd033ecc55b8b5d.png]: https://img-blog.csdnimg.cn/ec38ba035b5e40cc9dd033ecc55b8b5d.png
[dd11e38fa3704fa2ae178471ae310e13.png]: https://img-blog.csdnimg.cn/dd11e38fa3704fa2ae178471ae310e13.png
[a76fa2e02cbe492ea428f41737450b0f.png]: https://img-blog.csdnimg.cn/a76fa2e02cbe492ea428f41737450b0f.png
[c3cee6c1fcb84fb69ff345ad8a5287db.png]: https://img-blog.csdnimg.cn/c3cee6c1fcb84fb69ff345ad8a5287db.png
[fb69ed53c3234273b86bcba2d9a74071.png]: https://img-blog.csdnimg.cn/fb69ed53c3234273b86bcba2d9a74071.png