rabbitMQ(三) 不同的交换机规则

墨蓝 2022-06-02 05:09 141阅读 0赞

在上一篇的内容中，讲到rabbitMQ的发布和订阅，也就是使用了交换机里面的其中一个机制。  
现在继续接着讲其他的  
4）路由  
之前的发布和订阅那是全部发过去的，现在需要做的是给他需要的消息，其他不需要的就不接受。  
绑定是交换和队列之间的关系。这可以简单地理解为：队列对来自这个交换的消息感兴趣。  
绑定可以采用额外的routingKey参数。为了避免混淆basic\_publish参数，我们将其称为 绑定键。这是我们如何创建一个关键的绑定：

channel.queueBind(queue, EXCHANGE_NAME, "orange");

![这里写图片描述][SouthEast]  
实现上图代码  
EmitLogDirect.java:

import com.rabbitmq.client.BuiltinExchangeType;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    public class EmitLogDirect { 
        private  static  final String EXCHANGE_NAME = "direct_logs" ;
        //路由关键字
        private static  final  String[] routeKey = {
       "orange","black","green","white","red"};
        public static void main(String[] args) throws Exception {
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("localhost");
            Connection connection = connectionFactory.newConnection();
            Channel channel = connection.createChannel();
            //声明交换机
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
            for (String key : routeKey) {
                String string = "路由发送的信息："+key;
                channel.basicPublish(EXCHANGE_NAME, key, null, string.getBytes());
                System.out.println("[s] 发送的信息：{}"+string);
            }
            channel.close();
            connection.close();
        }
    }

ReceiveLogsDirect1.java

import java.io.IOException;
    
    import com.rabbitmq.client.AMQP.BasicProperties;
    import com.rabbitmq.client.BuiltinExchangeType;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    import com.rabbitmq.client.Consumer;
    import com.rabbitmq.client.DefaultConsumer;
    import com.rabbitmq.client.Envelope;
    
    public class ReceiveLogsDirect1 {
        private static final String EXCHANGE_NAME ="direct_logs";
        //
        private  static final String[] routekey = {
       "black","green"};
        public static void main(String[] args) throws Exception {
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("localhost");
            Connection connection = connectionFactory.newConnection();
            Channel channel = connection.createChannel();
            //声明交换器
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
            //获取匿名队列名称
            String queue = channel.queueDeclare().getQueue();
            //根据路由关键字进行绑定
            for (String string : routekey) {
                channel.queueBind(queue, EXCHANGE_NAME, string);
                System.out.println("路由关键字匹配------"+string+"-----成功");
            }
            Consumer consumer = new DefaultConsumer(channel){
    
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope,
                        BasicProperties properties, byte[] body) throws IOException {
                    System.out.println("ReceiveLogsDirect1 接受到的内容"+new String(body, "utf-8"));
                }
    
            };
            channel.basicConsume(queue, true, consumer);
        }
    }

ReceiveLogsDirect2.java

import java.io.IOException;
    
    import com.rabbitmq.client.AMQP.BasicProperties;
    import com.rabbitmq.client.BuiltinExchangeType;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    import com.rabbitmq.client.Consumer;
    import com.rabbitmq.client.DefaultConsumer;
    import com.rabbitmq.client.Envelope;
    
    public class ReceiveLogsDirect2 {
        private static final String EXCHANGE_NAME ="direct_logs";
        //
        private  static final String[] routekey = {
       "orange"};
        public static void main(String[] args) throws Exception {
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("localhost");
            Connection connection = connectionFactory.newConnection();
            Channel channel = connection.createChannel();
            //声明交换器
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
            //获取匿名队列名称
            String queue = channel.queueDeclare().getQueue();
            //根据路由关键字进行绑定
            for (String string : routekey) {
                channel.queueBind(queue, EXCHANGE_NAME, string);
                System.out.println("路由关键字匹配------"+string+"-----成功");
            }
            Consumer consumer = new DefaultConsumer(channel){
    
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope,
                        BasicProperties properties, byte[] body) throws IOException {
                    System.out.println("ReceiveLogsDirect2 接受到的内容"+new String(body, "utf-8"));
                }
    
            };
            channel.basicConsume(queue, true, consumer);
        }
    }

结果：  
![这里写图片描述][SouthEast 1]

5）主题  
尽管使用直接交换改进了我们的系统，但它仍然有局限性 - 它不能根据多个标准进行路由选择。  
发送到主题交换的消息不能有任意的 routing\_key - 它必须是由点分隔的单词列表。这些单词可以是任何东西，但通常它们指定连接到消息的一些功能。一些有效的路由键例子：“ stock.usd.nyse ”，“ nyse.vmw ”，“ quick.orange.rabbit ”。在路由选择键中可以有任意数量的字，最多255个字节。

绑定键也必须是相同的形式。主题交换背后的逻辑 类似于直接的 - 使用特定的路由密钥发送的消息将被传送到与匹配的绑定密钥绑定的所有队列。但是绑定键有两个重要的特殊情况：

\*（星号）可以代替一个字。  
＃（散列）可以代替零个或多个单词。  
![这里写图片描述][SouthEast 2]

EmitLogTopic.java

import com.rabbitmq.client.BuiltinExchangeType;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    public class EmitLogTopic {
        private static final String EXCHANGE_NAME = "topic_logs";
        private static final String[] routeKeys={
            "quick.orange.rabbit",
            "lazy.orange.elephant",
            "quick.orange.fox",
            "lazy.brown.fox",
            "quick.brown.fox",
            "quick.orange.male.rabbit",
            "lazy.orange.male.rabbit"};
        public static void main(String[] args) throws Exception {
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("localhost");
            Connection connection = connectionFactory.newConnection();
            Channel channel = connection.createChannel();
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
            for (String string : routeKeys) {
                String messageString  = "From "+string + "of message";
                channel.basicPublish(EXCHANGE_NAME, string, null, messageString.getBytes());
                System.out.println(messageString);
            }
            channel.close();
            connection.close();
        }
    }

ReceiveLogsTopic1.java

import java.io.IOException;
    
    import com.rabbitmq.client.AMQP.BasicProperties;
    import com.rabbitmq.client.BuiltinExchangeType;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    import com.rabbitmq.client.Consumer;
    import com.rabbitmq.client.DefaultConsumer;
    import com.rabbitmq.client.Envelope;
    
    public class ReceiveLogsTopic1 {
        private static final String EXCHANGE_NAME = "topic_logs";
        private static final String[] routekey = {
       "*.orange.*"};
        public static void main(String[] args) throws Exception{
            ConnectionFactory connectionFactory  = new ConnectionFactory();
            connectionFactory.setHost("localhost");
            Connection connection = connectionFactory.newConnection();
            Channel channel = connection.createChannel();
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
            String queue = channel.queueDeclare().getQueue();
            for (String string : routekey) {
                channel.queueBind(queue, EXCHANGE_NAME, string);
            }
            Consumer consumer = new DefaultConsumer(channel){
    
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope,
                        BasicProperties properties, byte[] body) throws IOException {
                    System.out.println("ReceiveLogsTopic1 接受的信息"+new String(body, "utf-8"));
                }
    
            };
            channel.basicConsume(queue,true, consumer);
        }
    }

ReceiveLogsTopic2.java

import java.io.IOException;
    
    import com.rabbitmq.client.AMQP.BasicProperties;
    import com.rabbitmq.client.BuiltinExchangeType;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    import com.rabbitmq.client.Consumer;
    import com.rabbitmq.client.DefaultConsumer;
    import com.rabbitmq.client.Envelope;
    
    public class ReceiveLogsTopic2 {
        private static final String EXCHANGE_NAME = "topic_logs";
        private static final String[] routekey = {
       "*.*.rabbit", "lazy.#"};
        public static void main(String[] args) throws Exception{
            ConnectionFactory connectionFactory  = new ConnectionFactory();
            connectionFactory.setHost("localhost");
            Connection connection = connectionFactory.newConnection();
            Channel channel = connection.createChannel();
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
            String queue = channel.queueDeclare().getQueue();
            for (String string : routekey) {
                channel.queueBind(queue, EXCHANGE_NAME, string);
            }
            Consumer consumer = new DefaultConsumer(channel){
    
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope,
                        BasicProperties properties, byte[] body) throws IOException {
                    System.out.println("ReceiveLogsTopic2 接受的信息"+new String(body, "utf-8"));
                }
    
            };
            channel.basicConsume(queue,true, consumer);
        }
    }

结果：  
![这里写图片描述][SouthEast 3]

[SouthEast]: /images/20220602/c0af883655ed4af78422ad265b2c3dfb.png
[SouthEast 1]: /images/20220602/43c4ab9f500d40b9acfc0b334cb52181.png
[SouthEast 2]: /images/20220602/f559b945762b4331be652390255de699.png
[SouthEast 3]: /images/20220602/f2535a4db5704d4386e0da69e302d877.png