RabbitMQ Routing（路由）工作模式

兰舟千帆

发布于 2022-08-03 18:13:57

4470

发布于 2022-08-03 18:13:57

文章被收录于专栏：兰舟千帆的java学习笔记兰舟千帆的java学习笔记

RabbitMQ Routing（路由）工作模式

路由的这种模式和发布订阅的模式相比多了一个路由的环节。要求队列在绑定在绑定到交换机的时候指定到对应的路由。路由的作用是什么呢？作为一种分发的规则。可以按照相应的条件指定和分发。

官方还是很形象的给出了我们的模型图。

这里对日志的级别采用了不同的处理方式。分别是error,info,warninng。其实日志移动有五种这里给出了三种日志举例的处理方式。

DEBUG：详细的信息,通常只出现在诊断问题上 INFO：确认一切按预期运行 WARNING：一个迹象表明,一些意想不到的事情发生了,或表明一些问题在不久的将来(例如。磁盘空间低”)。这个软件还能按预期工作。 ERROR：更严重的问题,软件没能执行一些功能 CRITICAL：一个严重的错误,这表明程序本身可能无法继续运行

运行出现的日志info,error,warning会路由到一个队列，然后单独的error也会路由到一个队列。

其实还是比较简单的，我们来看具体的代码实现。

那我们还需要一个生产者首先，来看代码。

package com.jgdabc.producer;

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_Routing {
    public static void main(String[] args) throws IOException, TimeoutException {

//       1 创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        //       2 设置连接参数
        connectionFactory.setHost("123.56.4.32"); //服务器主机ip公网 默认localhost
        connectionFactory.setPort(5672); //消息端口 默认5672
        connectionFactory.setVirtualHost("/shabi");//虚拟机 默认/
        connectionFactory.setUsername("jgdabc");//用户名 默认guest
        connectionFactory.setPassword("123456");//密码 默认guest

//        3创建连接connection
        Connection connection = connectionFactory.newConnection();
//       4 创建Channel
        Channel channel = connection.createChannel();
//        5：创建交换机
//        exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments)
//      1:exchange : 交换机名称
//       2: type:交换机的类型 BuiltinExchangeType代表枚举类型
//        Direct("direct"):定向
//        FANOUT("fanout") : 扇形广播，发送消息到每一个与之绑定的队列
//        TOPIC("topic") 通配符的方式
//        HEADERS("headers")
//        3:durable是否持久化
//        4：autoDelete:是否自动删除
//        5：internal 内部使用
//        6Larguments:参数列表
        String exchangeName = "test_direct";

        channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT,true,false,false,null);
//        6：创建队列
        String queueName = "test_direct_queue";
        String queueName01 = "test_direct_queue01";
        channel.queueDeclare(queueName,true,false,false,null);
        channel.queueDeclare(queueName01,true,false,false,null);

//          queueBind(String queue, String exchange, String routingKey)
//        7 绑定队列和交换机
//        queue:绑定的队列名称
//        exchange:交换机名称
//        routingKey:路由键，绑定规则
//        如果倦鸟还击的类型为fanout,routingKey设置为“”
         channel.queueBind(queueName,exchangeName,"error");
         channel.queueBind(queueName,exchangeName,"info");
         channel.queueBind(queueName,exchangeName,"warning");

         channel.queueBind(queueName01,exchangeName,"error");
//        8:发送消息
        String body = "日志信息。张三调用了findAll方法,日志级别info....";
        channel.basicPublish(exchangeName,"info",null,body.getBytes());

//        9：释放资源
        channel.close();
        connection.close();


    }
}

说明一次处理逻辑，我们这回的交换机需要指定为DIRECT，也就是一种定向的模式，很明显我们是定向分配的，如果你要是广播的模式的话，那么路由就不能做成定向的。这是一点。

抽出来说明一些，下面这段是创建交换机以及创建出队列

        String exchangeName = "test_direct";

        channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT,true,false,false,null);
//        6：创建队列
        String queueName = "test_direct_queue";
        String queueName01 = "test_direct_queue01";
        channel.queueDeclare(queueName,true,false,false,null);
        channel.queueDeclare(queueName01,true,false,false,null);

下面这段是进行一个路由的绑定

 channel.queueBind(queueName,exchangeName,"error");
 channel.queueBind(queueName,exchangeName,"info");
 channel.queueBind(queueName,exchangeName,"warning");
 channel.queueBind(queueName01,exchangeName,"error");

下面是我们发送消息，body是要发送的消息。这里指定了交换机和路由，这是一个info级别的日志，所以不会发送到单独绑定error的队列。当然你在这里可以更改。只是举个例子。

 String body = "日志信息。张三调用了findAll方法,日志级别info....";
 channel.basicPublish(exchangeName,"info",null,body.getBytes());

然后呢，然后来看消费者的代码，我们只说明一个，另一个是一样的道理。

消费者一：

package com.jgdabc.consumer;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Consumer_Routing {
    public static void main(String[] args) throws IOException, TimeoutException {

//       1 创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        //       2 设置连接参数
        connectionFactory.setHost("123.56.4.32"); //服务器主机ip公网 默认localhost
        connectionFactory.setPort(5672); //消息端口 默认5672
        connectionFactory.setVirtualHost("/shabi");//虚拟机 默认/
        connectionFactory.setUsername("jgdabc");//用户名 默认guest
        connectionFactory.setPassword("123456");//密码 默认guest

//        3创建连接connection
        Connection connection = connectionFactory.newConnection();
//       4 创建Channel
        Channel channel = connection.createChannel();
        String queueName = "test_direct_queue";
        String queueName01 = "test_direct_queue01";

        channel.queueDeclare(queueName,true,false,false,null);
        channel.queueDeclare(queueName01,true,false,false,null);


//        发送消息

//5创建队列
//        queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<Stri
//参数说明: queue :队列名称
//        durable : 是否持久化 :当mq 重启数据还在
//        exclusive : 是否独占，只能有一个消费者监听这队列
//                      当connection关闭时候，是否删除队列
//        autoDelete:是否自动删除，当没有Consumer时候，是否自动删除

//         basicConsume(String queue, DeliverCallback deliverCallback, ConsumerShutdownSignalCallback shutdownSignalCallback)
//        参数说明
//        queue: 队列名称
//        autoAck : 是否自动确认
//        callback: 回调函数
        DefaultConsumer consumer = new DefaultConsumer(channel) {
//            回调方法，当收到消息后会自动执行该方法
//            consumerTag:消息表示
//            ebvelop:获取一些信息，交换机的信息，路由等等
//            properties:配置信息
//            body:数据
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                super.handleDelivery(consumerTag, envelope, properties, body);
//                System.out.println("consumerTag:"+consumerTag);
//                System.out.println("Exchange:"+envelope.getExchange());
//                System.out.println("RoutingKey:"+envelope.getRoutingKey());
//                System.out.println("properties:"+properties);
                  System.out.println("body:"+new String(body));
                  System.out.println("将日志信息打印到控制台。。。");
            }
        };
        channel.basicConsume(queueName,true,consumer);
//
    }
}

消费者二：

package com.jgdabc.consumer;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Consumer_Routing1 {
    public static void main(String[] args) throws IOException, TimeoutException {

//       1 创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        //       2 设置连接参数
        connectionFactory.setHost("123.56.4.32"); //服务器主机ip公网 默认localhost
        connectionFactory.setPort(5672); //消息端口 默认5672
        connectionFactory.setVirtualHost("/shabi");//虚拟机 默认/
        connectionFactory.setUsername("jgdabc");//用户名 默认guest
        connectionFactory.setPassword("123456");//密码 默认guest

//        3创建连接connection
        Connection connection = connectionFactory.newConnection();
//       4 创建Channel
        Channel channel = connection.createChannel();
        String queueName = "test_direct_queue";
        String queueName01 = "test_direct_queue01";

//        发送消息

//5创建队列
//        queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<Stri
//参数说明: queue :队列名称
//        durable : 是否持久化 :当mq 重启数据还在
//        exclusive : 是否独占，只能有一个消费者监听这队列
//                      当connection关闭时候，是否删除队列
//        autoDelete:是否自动删除，当没有Consumer时候，是否自动删除

        channel.queueDeclare(queueName,true,false,false,null);
        channel.queueDeclare(queueName01,true,false,false,null);
//        接收消息
//         basicConsume(String queue, DeliverCallback deliverCallback, ConsumerShutdownSignalCallback shutdownSignalCallback)
//        参数说明
//        queue: 队列名称
//        autoAck : 是否自动确认
//        callback: 回调函数
        DefaultConsumer consumer = new DefaultConsumer(channel) {
//            回调方法，当收到消息后会自动执行该方法
//            consumerTag:消息表示
//            ebvelop:获取一些信息，交换机的信息，路由等等
//            properties:配置信息
//            body:数据
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                super.handleDelivery(consumerTag, envelope, properties, body);
//                System.out.println("consumerTag:"+consumerTag);
//                System.out.println("Exchange:"+envelope.getExchange());
//                System.out.println("RoutingKey:"+envelope.getRoutingKey());
//                System.out.println("properties:"+properties);
                  System.out.println("body:"+new String(body));
                  System.out.println("将信息存储到数据库。。。");
            }
        };
        channel.basicConsume(queueName01,true,consumer);
//
    }
}

取来一个说明。这里面有一些多余的代码啊!不知道你有没有发现。生产者做过的事情消费者没必要做了。队列都创建过了。这里这段在消费者里的代码可以删除掉，不过并不影响，

        channel.queueDeclare(queueName,true,false,false,null);
        channel.queueDeclare(queueName01,true,false,false,null);

这里是一个处理逻辑

这里代表监听的队列。

 channel.basicConsume(queueName,true,consumer);

测试

你看这里只有一个消费者获取到了消息

获取到消息的做出响应的逻辑处理，没获取到消息的则不会进行处理。

这就是路由。

本文参与腾讯云自媒体分享计划，分享自作者个人站点/博客。

原始发表：2022-07-18，如有侵权请联系 cloudcommunity@tencent.com 删除

debug

本文分享自作者个人站点/博客前往查看

如有侵权，请联系 cloudcommunity@tencent.com 删除。

本文参与腾讯云自媒体分享计划，欢迎热爱写作的你一起参与！

登录后参与评论

0 条评论

热度