《RabbitMQ篇》交换机基本概览

生产者都是把消息给交换机，由交换机分发给消息队列。

routingKey：路由键，也可称为绑定，是交换机和队列之间的桥梁，交换机会根据routingKey来把消息转发到对应的队列。

Fanout

不处理路由键。你只需要简单的将队列绑定到交换机上。一个发送到交换机的消息都会被转发到与该交换机绑定的所有队列上。

通用工具类：

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

public class RabbitMQUtil {
    public static Channel getChannel() throws Exception {
        //得到工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setPort(5672);
        factory.setUsername("guest");
        factory.setPassword("guest");
        //生成连接
        Connection connection = factory.newConnection();
        //获取信道
        return connection.createChannel();
    }
}

生产者代码：创建一个Fanout Exchange。

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;

import java.util.Scanner;

public class FanoutProducer {

    private final static String EXCHANGE_NAME = "LogExchange";

    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMQUtil.getChannel();

        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);

        Scanner scanner = new Scanner(System.in);
        while (scanner.hasNext()) {
            String message = scanner.nextLine();
            channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
        }
    }
}

两个消费者代码：创建队列去绑定交换机

public class Consumer1 {
    private final static String FILE_QUEUE = "FileQueue";
    private final static String EXCHANGE_NAME = "LogExchange";

    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMQUtil.getChannel();
        channel.queueDeclare(FILE_QUEUE, false, false, false, null);
        channel.queueBind(FILE_QUEUE, EXCHANGE_NAME, "");

        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            System.out.printf("消息：%s,存储到磁盘%n", new String(delivery.getBody()));
        };
        channel.basicConsume(FILE_QUEUE, true, deliverCallback, consumerTag -> {});
    }
}

public class Consumer2 {
    private final static String CONSOLE_QUEUE = "ConsoleQueue";
    private final static String EXCHANGE_NAME = "LogExchange";

    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMQUtil.getChannel();

        channel.queueDeclare(CONSOLE_QUEUE, false, false, false, null);
        channel.queueBind(CONSOLE_QUEUE, EXCHANGE_NAME, "");

        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            System.out.printf("消息：%s,输出到控制台%n", new String(delivery.getBody()));
        };
        channel.basicConsume(CONSOLE_QUEUE, true, deliverCallback, consumerTag -> {});
    }
}

运行结果图如下。总结：在Fanout模式下，生产者生产消息发布到交换机后，交换机会把消息发送给所有与之绑定的队列，在该模式下，即使设置了routingKey也会无视它。

Direct

处理路由键。需要将一个队列绑定到交换机上，要求该消息与一个特定的路由键完全匹配。这是一个完整的匹配。

生产者代码：创建一个Direct交换机，发送消息到交换机同时指定routingKey。

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;

import java.util.Scanner;

public class DirectProducer {
    private final static String EXCHANGE_NAME = "LogExchange-Direct";

    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMQUtil.getChannel();

        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);

        Scanner scanner = new Scanner(System.in);
        while (scanner.hasNext()) {
            String message = scanner.nextLine();
            channel.basicPublish(EXCHANGE_NAME, "consoleRouteKey", null, message.getBytes());
        }
    }
}

消费者代码：让队列和交换机绑定，同时设置routingKey。

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;

public class Consumer1 {
    private final static String FILE_QUEUE = "FileQueue";
    private final static String EXCHANGE_NAME = "LogExchange-Direct";


    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMQUtil.getChannel();
        channel.queueDeclare(FILE_QUEUE, false, false, false, null);
        channel.queueBind(FILE_QUEUE, EXCHANGE_NAME, "fileRouteKey");

        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            System.out.printf("消息：%s,存储到磁盘%n", new String(delivery.getBody()));
        };
        channel.basicConsume(FILE_QUEUE, true, deliverCallback, consumerTag -> {});
    }
}

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;

public class Consumer2 {
    private final static String CONSOLE_QUEUE = "ConsoleQueue";
    private final static String EXCHANGE_NAME = "LogExchange-Direct";


    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMQUtil.getChannel();

        channel.queueDeclare(CONSOLE_QUEUE, false, false, false, null);
        channel.queueBind(CONSOLE_QUEUE, EXCHANGE_NAME, "consoleRouteKey");

        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            System.out.printf("消息：%s,输出到控制台%n", new String(delivery.getBody()));
        };
        channel.basicConsume(CONSOLE_QUEUE, true, deliverCallback, consumerTag -> {});
    }
}

执行结果如图，总结：在Direct模式下，交换机会根据routingKey把消息给相关的队列，如果没有这样的队列，消息会被丢弃。

Topic

给定的routingKey，与交换机和队列之间设置的routingKey 根据模式匹配（类似正则）转发到对应的队列。

该模式下的routingKey的命名是一个单词列表，以点号分隔开。例如：“log.error.console”。其中可以用 *(星号)可以代替一个位置，#(井号)可以替代零个或多个位置。

比如：现在队列Q1和交换机之间设置的routingKey为 *.orange.*

队列Q2和交换机之间设置的routingKey为 *.*.rabbit*、lazy.#

例如	说明
quick.orange.rabbit	被队列 Q1Q2 接收到
azy.orange.elephant	被队列 Q1Q2 接收到
quick.orange.fox	被队列 Q1 接收到
lazy.brown.fox	被队列 Q2 接收到
lazy.pink.rabbit	虽然满足两个绑定但只被队列 Q2 接收一次
quick.brown.fox	不匹配任何绑定不会被任何队列接收到会被丢弃
quick.orange.male.rabbit	是四个单词不匹配任何绑定会被丢弃
lazy.orange.male.rabbit	是四个单词但匹配 Q2

实际测试代码

import com.rabbitmq.client.Channel;

import java.util.HashMap;
import java.util.Map;

public class TopicProducer {
    private final static String EXCHANGE_NAME = "LogExchange-Topic";

    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMQUtil.getChannel();

        HashMap<String, String> bindingKeyMap = new HashMap<>();
        bindingKeyMap.put("quick.orange.rabbit", "1.被队列 Q1Q2 接收到");
        bindingKeyMap.put("lazy.orange.elephant", "2.被队列 Q1Q2 接收到");
        bindingKeyMap.put("quick.orange.fox", "3.被队列 Q1 接收到");
        bindingKeyMap.put("lazy.brown.fox", "4.被队列 Q2 接收到");
        bindingKeyMap.put("lazy.pink.rabbit", "5.虽然满足两个绑定但只被队列 Q2 接收一次");
        bindingKeyMap.put("quick.brown.fox", "6.不匹配任何绑定不会被任何队列接收到会被丢弃");
        bindingKeyMap.put("quick.orange.male.rabbit", "7.是四个单词不匹配任何绑定会被丢弃");
        bindingKeyMap.put("lazy.orange.male.rabbit", "8.是四个单词但匹配 Q2");
        for (Map.Entry<String,String> bindingKeyEntry : bindingKeyMap.entrySet()){
            String routingKey = bindingKeyEntry.getKey();
            String message = bindingKeyEntry.getValue();
            channel.basicPublish(EXCHANGE_NAME,routingKey,null,message.getBytes("UTF-8"));
            System.out.println("生产者发出消息:"+message);
        }
    }
}

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;

public class Consumer1 {
    private final static String FILE_QUEUE = "FileQueue";
    private final static String EXCHANGE_NAME = "LogExchange-Topic";


    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMQUtil.getChannel();
        channel.queueDeclare(FILE_QUEUE, false, false, false, null);
        channel.queueBind(FILE_QUEUE, EXCHANGE_NAME, "*.orange.*");

        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            System.out.printf("消息：%s,存储到磁盘%n", new String(delivery.getBody()));
        };
        channel.basicConsume(FILE_QUEUE, true, deliverCallback, consumerTag -> {});
    }
}

public class Consumer2 {
    private final static String CONSOLE_QUEUE = "ConsoleQueue";
    private final static String EXCHANGE_NAME = "LogExchange-Topic";


    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMQUtil.getChannel();

        channel.queueDeclare(CONSOLE_QUEUE, false, false, false, null);
        channel.queueBind(CONSOLE_QUEUE, EXCHANGE_NAME, "*.*.rabbit");
        channel.queueBind(CONSOLE_QUEUE, EXCHANGE_NAME, "lazy.#");

        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            System.out.printf("消息：%s,输出到控制台%n", new String(delivery.getBody()));
        };
        channel.basicConsume(CONSOLE_QUEUE, true, deliverCallback, consumerTag -> {});
    }
}

Header

不处理路由键。而是根据发送的消息内容中的headers属性进行匹配。在绑定 Queue 与 Exchange 时指定一组键值对；当消息发送到RabbitMQ 时会取到该消息的 headers 与 Exchange 绑定时指定的键值对进行匹配；如果完全匹配则消息会路由到该队列，否则不会路由到该队列。headers 属性是一个键值对，可以是 Hashtable，键值对的值可以是任何类型。而 fanout，direct，topic 的路由键都需要要字符串形式的。

匹配规则 x-match 有下列两种类型：

x-match = all ：表示所有的键值对都匹配才能接受到消息

x-match = any ：表示只要有键值对匹配就能接受到消息