🍁博客主页：👉不会压弯的小飞侠
✨欢迎关注：👉点赞👍收藏⭐留言✒
✨系列专栏：👉Linux专栏
🔥欢迎大佬指正，一起学习！一起加油！

---

🍁模式说明

• Routingkey 一般都是有一个或多个单词组成，多个单词之间以”.”分割。
• Topic主题模式可以实现 Publish/Subscribe发布与订阅模式 和 Routing路由模式 的功能；只是Topic在配置routing key 的时候可以使用通配符，显得更加灵活。
• 需要设置类型为topic的交换机，交换机和队列进行绑定，并且指定通配符方式的routing key，当发送消息到交换机后，交换机会根据routing key将消息发送到对应的队列。
• 通配符规则：
  • #：匹配零个或多个词
  • *：匹配不多不少恰好1个词

🍁通配符模式完成消息传递

• 编写生产者发送消息
  • 编写消息生产者 Producter

class Producer {
    public static void main(String[] args) throws Exception {
        //创建连接
        Connection connection = ConnectionUtil.getConnection();
        //创建信道
        Channel channel = connection.createChannel();
        String exchangeName = "test_topic";
        //声明交换机
        channel.exchangeDeclare(exchangeName, BuiltinExchangeType.TOPIC,true,false,false,null);
        String queue1Name = "test_topic_queue1";
        String queue2Name = "test_topic_queue2";
        //创建队列
        channel.queueDeclare(queue1Name,true,false,false,null);
        channel.queueDeclare(queue2Name,true,false,false,null);
        // 绑定队列和交换机
        /**
         *  参数：
         1. queue：队列名称
         2. exchange：交换机名称
         3. routingKey：路由键,绑定规则
         如果交换机的类型为fanout ,routingKey设置为""
         */
        // routing key  系统的名称.日志的级别。
        //需求： 所有error级别的日志存入数据库,所有order系统的日志存入数据库
        channel.queueBind(queue1Name,exchangeName,"#.xfx");
        channel.queueBind(queue1Name,exchangeName,"mks.*");
        channel.queueBind(queue1Name,exchangeName,"ls.#");
        channel.queueBind(queue2Name,exchangeName,"ok.*");
        String body = "该消息会进入队列一和队列二！！！";
        //发送消息
        //channel.basicPublish(exchangeName,"a.b.c.d.xfx",null,body.getBytes());
        //channel.basicPublish(exchangeName,"mks.ok.ok",null,body.getBytes());
        channel.basicPublish(exchangeName,"ls.a",null,body.getBytes());
        //channel.basicPublish(exchangeName,"ok.hi",null,body.getBytes());

        channel.close();
        connection.close();
    }
}

• 测试
  

• 编写消费者接收消息

  • 编写消息消费者Consumer1

public class Consumer1 {
    public static void main(String[] args) throws Exception {
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel();
        String queue1Name = "test_topic_queue1";
        Consumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("body："+new String(body));
            }
        };
        channel.basicConsume(queue1Name,true,consumer);
    }
}

• 编写消费者接收消息
  • 编写消息消费者Consumer2

public class Consumer2 {
    public static void main(String[] args) throws Exception {
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel();
        String queue2Name = "test_topic_queue2";
        Consumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("body："+new String(body));
            }
        };
        channel.basicConsume(queue2Name,true,consumer);
    }
}

• 测试
• 启动所有消费者，然后使用生产者发送消息；在消费者对应的控制台可以查看到生产者发送对应routing key对应队列的消息；到达按照需要接收的效果；并且这些routing key可以使用通配符。