rabbitmq中exchange交换机的四种类型

客官°小女子只卖身不卖艺 2022-10-10 09:41 184阅读 0赞

exchange有以下四种类型：

*  **direct**
 *  **fanout**
 *  **topic**
 *  **headers**

见枚举类**com.rabbitmq.client.BuiltinExchangeType**：

public enum BuiltinExchangeType { 
    
        DIRECT("direct"), FANOUT("fanout"), TOPIC("topic"), HEADERS("headers");
    ... ...

# **fanout** #

消息广播到绑定的队列，不管队列绑定了什么路由键，消息经过交换器，每个队列都有一份。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center]

## FanoutProducer ##

package com.morris.rabbit.exchange.fanout;
    
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    import java.util.stream.IntStream;
    
    public class FanoutProducer { 
    
        public final static String EXCHANGE_NAME = "fanout-exchange";
    
        public static void main(String[] args) throws IOException, TimeoutException { 
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("192.168.80.205");
            factory.setPort(5672);
            factory.setUsername("root");
            factory.setPassword("root");
            try (Connection connection = factory.newConnection();
                 Channel channel = connection.createChannel()) { 
    
                channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
                String message = "task";
    
                channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
                System.out.println(" [x] Sent '" + message + "'");
            }
        }
    }

## FanoutConsumer ##

package com.morris.rabbit.exchange.fanout;
    
    import com.rabbitmq.client.*;
    
    import static com.morris.rabbit.exchange.fanout.FanoutProducer.EXCHANGE_NAME;
    
    public class FanoutConsumer { 
    
        private final static String QUEUE_NAME = "fanout-queue";
    
        public static void main(String[] argv) throws Exception { 
    
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("192.168.80.205");
            factory.setPort(5672);
            factory.setUsername("root");
            factory.setPassword("root");
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "*");
            System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
    
            DeliverCallback deliverCallback = (consumerTag, delivery) -> { 
                String message = new String(delivery.getBody(), "UTF-8");
                System.out.println(" [x] Received '" + message + "'");
            };
            channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { });
        }
    }

## FanoutConsumer2 ##

package com.morris.rabbit.exchange.fanout;
    
    import com.rabbitmq.client.*;
    
    import static com.morris.rabbit.exchange.fanout.FanoutProducer.EXCHANGE_NAME;
    
    public class FanoutConsumer2 { 
    
        private final static String QUEUE_NAME = "fanout-queue2";
    
        public static void main(String[] argv) throws Exception { 
    
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("192.168.80.205");
            factory.setPort(5672);
            factory.setUsername("root");
            factory.setPassword("root");
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            channel.queueBind(QUEUE_NAME, FanoutProducer.EXCHANGE_NAME, "*");
            System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
    
            DeliverCallback deliverCallback = (consumerTag, delivery) -> { 
                String message = new String(delivery.getBody(), "UTF-8");
                System.out.println(" [x] Received '" + message + "'");
            };
            channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { });
        }
    }

## 测试 ##

先运行FanoutConsumer和FanoutConsumer2，再运行FanoutProducer。

运行结果如下：

FanoutProducer：

[x] Sent 'task'

FanoutConsumer：

[*] Waiting for messages. To exit press CTRL+C
     [x] Received 'task'

FanoutConsumer2：

[*] Waiting for messages. To exit press CTRL+C
     [x] Received 'task'

# direct #

路由键完全匹配，消息被投递到对应的队列，direct交换器是默认交换器。声明一个队列时，会自动绑定到默认交换器，并且以队列名称作为路由  
键。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center 1]

## DirectProducer ##

package com.morris.rabbit.exchange.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 DirectProducer { 
    
        public final static String EXCHANGE_NAME = "direct-exchange";
    
        public static void main(String[] args) throws IOException, TimeoutException { 
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("192.168.80.205");
            factory.setPort(5672);
            factory.setUsername("root");
            factory.setPassword("root");
            try (Connection connection = factory.newConnection();
                 Channel channel = connection.createChannel()) { 
                channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
    
                channel.basicPublish(EXCHANGE_NAME, "c1", null, "task1".getBytes());
                System.out.println(" [x] Sent 'task1'");
    
                channel.basicPublish(EXCHANGE_NAME, "c2", null, "task2".getBytes());
                System.out.println(" [x] Sent 'task2'");
            }
        }
    }

## DirectConsumer ##

package com.morris.rabbit.exchange.direct;
    
    import com.rabbitmq.client.*;
    
    import static com.morris.rabbit.exchange.direct.DirectProducer.EXCHANGE_NAME;
    
    public class DirectConsumer { 
    
        private final static String QUEUE_NAME = "direct-queue";
    
        public static void main(String[] argv) throws Exception { 
    
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("192.168.80.205");
            factory.setPort(5672);
            factory.setUsername("root");
            factory.setPassword("root");
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "c1");
            System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
    
            DeliverCallback deliverCallback = (consumerTag, delivery) -> { 
                String message = new String(delivery.getBody(), "UTF-8");
                System.out.println(" [x] Received '" + message + "'");
            };
            channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { });
        }
    }

## DirectConsumer2 ##

package com.morris.rabbit.exchange.direct;
    
    import com.rabbitmq.client.*;
    
    import static com.morris.rabbit.exchange.direct.DirectProducer.EXCHANGE_NAME;
    
    public class DirectConsumer2 { 
    
        private final static String QUEUE_NAME = "direct-queue2";
    
        public static void main(String[] argv) throws Exception { 
    
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("192.168.80.205");
            factory.setPort(5672);
            factory.setUsername("root");
            factory.setPassword("root");
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            channel.queueBind(QUEUE_NAME, DirectProducer.EXCHANGE_NAME, "c2");
            System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
    
            DeliverCallback deliverCallback = (consumerTag, delivery) -> { 
                String message = new String(delivery.getBody(), "UTF-8");
                System.out.println(" [x] Received '" + message + "'");
            };
            channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { });
        }
    }

## 测试 ##

先运行DirectConsumer和DirectConsumer2 ，再运行DirectProducer。

运行结果如下：

DirectProducer：

[x] Sent 'task1'
     [x] Sent 'task2'

DirectConsumer：

[*] Waiting for messages. To exit press CTRL+C
     [x] Received 'task1'

DirectConsumer2：

[*] Waiting for messages. To exit press CTRL+C
     [x] Received 'task2'

# topic #

通过使用“*”和“\#”通配符进行处理，使来自不同源头的消息到达同一个队列，”.”将路由键分为了几个标识符，“*”匹配 1 个，“\#”匹配一个  
或多个。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center 2]

## TopicProducer ##

package com.morris.rabbit.exchange.topic;
    
    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 TopicProducer { 
    
        public final static String EXCHANGE_NAME = "topic-exchange";
    
        public static void main(String[] args) throws IOException, TimeoutException { 
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("192.168.80.205");
            factory.setPort(5672);
            factory.setUsername("root");
            factory.setPassword("root");
            try (Connection connection = factory.newConnection();
                 Channel channel = connection.createChannel()) { 
                channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
    
                channel.basicPublish(EXCHANGE_NAME, "xxx.aaa", null, "task1".getBytes());
                System.out.println(" [x] Sent 'task1'");
    
                channel.basicPublish(EXCHANGE_NAME, "xxx.bbb", null, "task2".getBytes());
                System.out.println(" [x] Sent 'task2'");
            }
        }
    }

## TopicConsumer ##

package com.morris.rabbit.exchange.topic;
    
    import com.rabbitmq.client.*;
    
    import static com.morris.rabbit.exchange.topic.TopicProducer.EXCHANGE_NAME;
    
    public class TopicConsumer { 
    
        private final static String QUEUE_NAME = "topic-queue";
    
        public static void main(String[] argv) throws Exception { 
    
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("192.168.80.205");
            factory.setPort(5672);
            factory.setUsername("root");
            factory.setPassword("root");
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "xxx.#");
            System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
    
            DeliverCallback deliverCallback = (consumerTag, delivery) -> { 
                String message = new String(delivery.getBody(), "UTF-8");
                System.out.println(" [x] Received '" + message + "'");
            };
            channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { });
        }
    }

## TopicConsumer2 ##

package com.morris.rabbit.exchange.topic;
    
    import com.rabbitmq.client.*;
    
    import static com.morris.rabbit.exchange.topic.TopicProducer.EXCHANGE_NAME;
    
    public class TopicConsumer2 { 
    
        private final static String QUEUE_NAME = "topic-queue2";
    
        public static void main(String[] argv) throws Exception { 
    
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("192.168.80.205");
            factory.setPort(5672);
            factory.setUsername("root");
            factory.setPassword("root");
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            channel.queueBind(QUEUE_NAME, TopicProducer.EXCHANGE_NAME, "#.bbb");
            System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
    
            DeliverCallback deliverCallback = (consumerTag, delivery) -> { 
                String message = new String(delivery.getBody(), "UTF-8");
                System.out.println(" [x] Received '" + message + "'");
            };
            channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { });
        }
    }

## 测试 ##

先运行TopicConsumer和TopicConsumer2 ，再运行TopicProducer 。

运行结果如下：

TopicProducer ：

[x] Sent 'task1'
     [x] Sent 'task2'

TopicConsumer：

[*] Waiting for messages. To exit press CTRL+C
     [x] Received 'task1'
     [x] Received 'task2'

TopicConsumer2：

[*] Waiting for messages. To exit press CTRL+C
     [x] Received 'task2'

# headers #

header exchange和direct exchange有点相似，但是不同于direct exchange的路由是基于路由键，header exchange的路由值基于消息的header数据。

## HeaderProducer ##

package com.morris.rabbit.exchange.header;
    
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.HashMap;
    import java.util.Map;
    import java.util.concurrent.TimeoutException;
    
    public class HeaderProducer { 
    
        public final static String EXCHANGE_NAME = "header-exchange";
    
        public static void main(String[] args) throws IOException, TimeoutException { 
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("192.168.80.205");
            factory.setPort(5672);
            factory.setUsername("root");
            factory.setPassword("root");
            try (Connection connection = factory.newConnection();
                 Channel channel = connection.createChannel()) { 
                channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.HEADERS);
    
                Map<String,Object> headersMap  = new HashMap<>();
                headersMap.put("api-version", "5.1.0");
                headersMap.put("platform", "android");
                AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
                AMQP.BasicProperties properties = builder.headers(headersMap).build();
                channel.basicPublish(EXCHANGE_NAME, "", properties, "task1".getBytes());
                System.out.println(" [x] Sent 'task1'");
    
                headersMap.put("api-version", "5.2.0");
                headersMap.put("platform", "android");
                properties = builder.headers(headersMap).build();
                channel.basicPublish(EXCHANGE_NAME, "", properties, "task2".getBytes());
                System.out.println(" [x] Sent 'task2'");
    
            }
        }
    }

## HeaderConsumer ##

package com.morris.rabbit.exchange.header;
    
    import com.rabbitmq.client.*;
    
    import java.util.HashMap;
    import java.util.Map;
    
    import static com.morris.rabbit.exchange.header.HeaderProducer.EXCHANGE_NAME;
    
    public class HeaderConsumer { 
    
        private final static String QUEUE_NAME = "header-queue";
    
        public static void main(String[] argv) throws Exception { 
    
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("192.168.80.205");
            factory.setPort(5672);
            factory.setUsername("root");
            factory.setPassword("root");
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.HEADERS);
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            Map<String,Object> headersMap  = new HashMap<>();
            headersMap.put("api-version", "5.1.0");
            headersMap.put("platform", "android");
            headersMap.put("x-match","all"); // 全匹配
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "", headersMap);
            System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
    
            DeliverCallback deliverCallback = (consumerTag, delivery) -> { 
                String message = new String(delivery.getBody(), "UTF-8");
                System.out.println(" [x] Received '" + message + "'");
            };
            channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { });
        }
    }

## HeaderConsumer2 ##

package com.morris.rabbit.exchange.header;
    
    import com.rabbitmq.client.*;
    
    import java.util.HashMap;
    import java.util.Map;
    
    import static com.morris.rabbit.exchange.header.HeaderProducer.EXCHANGE_NAME;
    
    public class HeaderConsumer2 { 
    
        private final static String QUEUE_NAME = "header-queue2";
    
        public static void main(String[] argv) throws Exception { 
    
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("192.168.80.205");
            factory.setPort(5672);
            factory.setUsername("root");
            factory.setPassword("root");
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.HEADERS);
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            Map<String,Object> headersMap  = new HashMap<>();
            headersMap.put("api-version", "5.1.0");
            headersMap.put("platform", "android");
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "", headersMap);
            System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
    
            DeliverCallback deliverCallback = (consumerTag, delivery) -> { 
                String message = new String(delivery.getBody(), "UTF-8");
                System.out.println(" [x] Received '" + message + "'");
            };
            channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { });
        }
    }

## 测试 ##

先运行HeaderConsumer和HeaderConsumer2 ，再运行HeaderProducer 。

运行结果如下：

HeaderProducer ：

[x] Sent 'task1'
     [x] Sent 'task2'

HeaderConsumer：

[*] Waiting for messages. To exit press CTRL+C
     [x] Received 'task1'

HeaderConsumer2：

[*] Waiting for messages. To exit press CTRL+C
     [x] Received 'task1'
     [x] Received 'task2'

消息header数据里有一个特殊值**x-match**，它有两个值：

*  all：一个传送消息的header里的键值对和交换机的header键值对全部匹配，才可以路由到对应交换机  
    。
 *  any：默认值。一个传送消息的header里的键值对和交换机的header键值对任意一个匹配，就可以路由到对应交换机  
    。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center]: /images/20221005/c631a839a6034f6f903de49c123651a8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221005/244bb3ba875546c9b1a37d3f1ac528a9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221005/7233c1afe6bd4b35b083acfa350edb48.png