RabbitMQ消息传递模型的核心思想是：生产者生产的消息从来都不会直接发送到队列，相反，生产者只能将消息发送到交换机中。
相反，生产者只能将消息发送到交换机上，交换机再将消息推入到队列当中，由消费者进行消费。

交换机工作的内容非常简单，一方面它接收来自生产者的消息，另一方面将其推入到队列中。交换机必须确切知道如何处理收到的消息，是应该将其放入到特定的队列中，还是直接丢弃掉，这主要取决于交换机的类型。

交换机

交换机的类型

总共有如下几种类型：

Direct：直接 Topic：主题 Headers：标题 Fanout：扇出

无名交换机

在此之前，我们其实没有涉及（讨论）到与交换机有关的内容，但是我们却可以正常的使用消息队列。
这正是因为我们之前使用的都是默认交换机，一般使用空字符串进行标识

第一个参数是交换机的名称，空字符串表示默认或者无名称交换机；消息能够发送到队列中其实是由routingKey(bindingKey)所决定的

临时队列

根据前面的知识可以知道，如果我们再创建队列的时候没有指明要进行持久化，那么当我们关闭MQ的服务，队列就会被删除掉。
而假如我们每次连接到MQ服务的时候，如果需要一个全新的队列，那么就可以创建一个具有随机名称的队列，并当我们断开消费者的连接，队列就会被自动删除

绑定

生产者并不会直接跟队列相互发送消息交流，而是将消息发送到交换机，再由交换机将消息推入到特定的队列中。绑定Binding就是交换机和队列之间的桥梁，他可以告诉交换机和哪个队列之间有绑定关系。下图中，交换机X就是和队列Q1和Q2进行了绑定

Fanout类型

顾名思义，fanout类型是将接收到的所有消息广播到与其有关联的所有队列中。系统中有些默认交换机就是Fanout类型

实战

将类型为fanout的交换机绑定到如图所示的两个临时队列上，并发送广播消息

/**
 * Created By Intellij IDEA
 *
 * @author ssssheep
 * @package com.ctgu.sheep.test11
 * @datetime 2022/9/18 星期日
 */
public class LogEmit {
    public static final String EXCHANGE_NAME = "logs_test";
    public static void main(String[] args) throws Exception {
        Channel channel = MQUtil.getChannel();
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
        Scanner scanner = new Scanner(System.in);
        while (scanner.hasNext()) {
            String msg = scanner.nextLine();
            channel.basicPublish(EXCHANGE_NAME, "logs", null, msg.getBytes());
            System.out.println("生产者发送消息：" + msg);
        }
    }
}

/**
 * Created By Intellij IDEA
 *
 * @author ssssheep
 * @package com.ctgu.sheep.test11
 * @datetime 2022/9/18 星期日
 */
public class LogConsumer1 {
    public static void main(String[] args) throws Exception {
        for (int i = 1; i <= 2; i++) {
            MyThread thread = new MyThread(i + "");
            thread.start();
            System.out.println("消费者" + i + "已启动...");
        }
    }
}
class MyThread extends Thread {
    public static final String EXCHANGE_NAME = "logs_test";
    private final String tName;
    public MyThread(String tName) {
        this.tName = tName;
    }
    @Override
    public void run() {
        Channel channel = null;
        try {
            channel = MQUtil.getChannel();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        try {
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        String queue = null;
        try {
            queue = channel.queueDeclare().getQueue();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        try {
            channel.queueBind(queue, EXCHANGE_NAME, "logs");
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        try {
            channel.basicConsume(queue, true, (consumerTag, message) -> {
                System.out.println("消费者" + tName + "接收到消息：" + new String(message.getBody()));
            }, consumerTag -> {
            });
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
}

效果

两个消费者线程，可以同时从队列中获取到生产者发过来的数据

Direct类型

在上面广播类型中，只要队列和交换机进行了绑定，那么交换机就会将收到的消息转发到与之关联的所有队列中但是有时候我们并不想将消息进行这样无所谓的广播而是有选择的进行转发，如下图所示

在上面这张图中，我们可以看到 X 绑定了两个队列，绑定类型是 direct。队列Q1 绑定键为 orange，队列 Q2 绑定键有两个:一个绑定键为 black，另一个绑定键为 green. 在这种绑定情况下，生产者发布消息到 exchange 上，绑定键为 orange 的消息会被发布到队列 Q1。绑定键为 blackgreen 和的消息会被发布到队列 Q2，其他消息类型的消息将被丢弃。

一个队列可以绑定多个key，一个key也可以被多个队列进行绑定，如下图所示：

实战

通过用户的输入，将消息转发到对应的队列中

/**
 * Created By Intellij IDEA
 *
 * @author ssssheep
 * @package com.ctgu.sheep.test11
 * @datetime 2022/9/18 星期日
 */
public class LogEmit {
    public static final String EXCHANGE_NAME = "logs_test";
    public static void main(String[] args) throws Exception {
        Channel channel = MQUtil.getChannel();
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
        Scanner scanner = new Scanner(System.in);
        while (scanner.hasNext()) {
            String msg = scanner.nextLine();
            String[] strings = msg.split(":");
            channel.basicPublish(EXCHANGE_NAME, strings[0].trim(), null, strings[1].trim().getBytes());
            System.out.println("生产者发送消息：" + strings[1].trim());
        }
    }
}

/**
 * Created By Intellij IDEA
 *
 * @author ssssheep
 * @package com.ctgu.sheep.test11
 * @datetime 2022/9/18 星期日
 */
public class LogConsumer1 {
    public static void main(String[] args) throws Exception {
        String[] routingKeys = {"logs", "logs", "test", "test"};
        for (int i = 0; i < 4; i++) {
            MyThread thread = new MyThread(i + 1 + "", routingKeys[i]);
            thread.start();
            System.out.println("消费者" + (i + 1) + "已启动...");
        }
    }
}
class MyThread extends Thread {
    public static final String EXCHANGE_NAME = "logs_test";
    private final String tName;
    private final String bindingKey;
    public MyThread(String tName, String bindingKey) {
        this.tName = tName;
        this.bindingKey = bindingKey;
    }
    @Override
    public void run() {
        Channel channel = null;
        try {
            channel = MQUtil.getChannel();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        try {
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        String queue = null;
        try {
            queue = channel.queueDeclare().getQueue();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        try {
            channel.queueBind(queue, EXCHANGE_NAME, bindingKey);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        try {
            channel.basicConsume(queue, true, (consumerTag, message) -> {
                String routingKey = message.getEnvelope().getRoutingKey();
                System.out.println("消费者" + tName + "接收到消息：" + new String(message.getBody()));
                System.out.println("routing key is:" + routingKey);
            }, consumerTag -> {
            });
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
}

效果

Topics类型

在上一个小节中，我们改进了日志记录系统。我们没有使用只能进行随意广播的 fanout 交换机，而是
使用了 direct 交换机，从而有能实现有选择性地接收日志。尽管使用direct 交换机改进了我们的系统，但是它仍然存在局限性
比方说我们想接收的日志类型有 info.base 和 info.advantage，某个队列只想 info.base 的消息，那这个时候direct 就办不到了。这个时候就只能使用 topic 类型。

简单理解下，就是topic类型可以根据一定的规则进行匹配转发

发送到类型是topic交换机消息的routingKey不能随便写，必须是要满足一定的要求——必须是一个单词列表，并且以点号进行分割。
单词列表中单词可以是任意的，在这些单词列表中，有两个替换符号是需要注意的。

*：可以替换一个单词
#：可以替换零个、一个和多个单词

如上图所示，是一个队列绑定关系图，交换机转发的情况如下所示：

lazy.orange.male.rabbit：虽然是4个单词，但是#可以匹配多个单词，因此是可以匹配上Q2的

实战

/**
 * Created By Intellij IDEA
 *
 * @author ssssheep
 * @package com.ctgu.sheep.test9
 * @datetime 2022/9/17 星期六
 */
public class TopicLogEmit {
    public static final String EXCHANGE_NAME = "topic_logs";
    public static void main(String[] args) throws Exception {
        Channel channel = MQUtil.getChannel();
        channel.exchangeDeclare(EXCHANGE_NAME, "topic");
        HashMap<String, String> map = new HashMap<>();
        map.put("quick.orange.rabbit", "被队列Q1Q2接收到");
        map.put("lazy.orange.elephant", "被队列Q1Q2接收到");
        map.put("quick.orange.fox", "被队列Q1接收到");
        map.put("lazy.brown.fox", "被队列Q2接收到");
        map.put("lazy.pink.rabbit", "虽然满足两个绑定但只被队列Q2接收一次");
        map.put("quick.brown.fox", "不匹配任何绑定不会被任何队列接收到会被丢弃");
        map.put("quick.orange.male.rabbit", "是四个单词所以不会被匹配到");
        map.put("lazy.orange.male.rabbit", "是四个单词但是可以匹配Q2");
        map.forEach((key, value) -> {
            try {
                channel.basicPublish(EXCHANGE_NAME, key, null, value.getBytes("UTF-8"));
                System.out.println("生产者发送消息：" + value);
            } catch (IOException e) {
                e.printStackTrace();
                throw new RuntimeException(e);
            }
        });
    }
}

public class RecvTopLog1 {
    public static final String EXCHANGE_NAME = "topic_logs";
    public static void main(String[] args) throws Exception {
        Channel channel = MQUtil.getChannel();
        channel.exchangeDeclare(EXCHANGE_NAME, "topic");
        String queueName = channel.queueDeclare().getQueue();
        channel.queueBind(queueName, EXCHANGE_NAME, "*.orange.*");
        System.out.println("队列:" + queueName + "等待接收消息");
        channel.basicConsume(queueName, true, (consumerTag, message) -> {
            System.out.println("队列" + queueName + "接收到消息：" + new String(message.getBody(), "UTF-8"));
            System.out.println("绑定键为：" + message.getEnvelope().getRoutingKey());
        }, consumerTag -> {
        });
    }
}

public class RecvTopLog2 {
    public static final String EXCHANGE_NAME = "topic_logs";
    public static void main(String[] args) throws Exception {
        Channel channel = MQUtil.getChannel();
        channel.exchangeDeclare(EXCHANGE_NAME, "topic");
        String queueName = channel.queueDeclare().getQueue();
        channel.queueBind(queueName, EXCHANGE_NAME, "*.*.rabbit");
        channel.queueBind(queueName, EXCHANGE_NAME, "lazy.#");
        System.out.println("队列:" + queueName + "等待接收消息");
        channel.basicConsume(queueName, true, (consumerTag, message) -> {
            System.out.println("队列" + queueName + "接收到消息：" + new String(message.getBody(), "UTF-8"));
            System.out.println("绑定键为：" + message.getEnvelope().getRoutingKey());
        }, consumerTag -> {
        });
    }
}

效果

根据特定的规则可以正常收到消息