RabbitMQ工作模式之通配符模式詳解

更新時間：2026年01月15日 11:06:24 作者：新綠MEHO

通配符模式是一種靈活的消息傳遞模式,可以根據(jù)消息的路由鍵（routing key）和綁定（binding）模式來實現(xiàn)精確的消息過濾和匹配,本文給大家介紹RabbitMQ工作模式之通配符模式,感興趣的朋友跟隨小編一起看看吧

通配符模式

概述

通配符模式是一種靈活的消息傳遞模式，可以根據(jù)消息的路由鍵（routing key）和綁定（binding）模式來實現(xiàn)精確的消息過濾和匹配。在RabbitMQ中，路由鍵由生產(chǎn)者定義，用于標識消息的目的地；而綁定則由消費者定義，用于指定消息的接收規(guī)則。

路由模式的升級版, 在routingKey的基礎上,增加了通配符的功能, 使之更加靈活.

Topics和Routing的基本原理相同，即：?產(chǎn)者將消息發(fā)給交換機，交換機根據(jù)RoutingKey將消息轉(zhuǎn)發(fā)給與RoutingKey匹配的隊列. 類似于正則表達式的?式來定義Routingkey的模式.

不同之處是：routingKey的匹配?式不同，Routing模式是相等匹配，topics模式是通配符匹配.

應用場景

RabbitMQ的通配符模式在需要根據(jù)消息的特定屬性進行路由和過濾的場景中非常有用。例如，在一個日志系統(tǒng)中，可以使用通配符模式來將不同級別的日志消息路由到不同的隊列中，以便進行不同的處理和分析。

優(yōu)勢

通配符模式的優(yōu)勢在于它可以靈活地匹配消息，使得消息可以根據(jù)不同的條件進行過濾和選擇。通過合理地定義綁定和路由鍵，可以實現(xiàn)復雜的消息過濾和路由策略，提高系統(tǒng)的靈活性和性能。

代碼案例

引入依賴

<!-- https://mvnrepository.com/artifact/com.rabbitmq/amqp-client -->
<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>5.21.0</version>
</dependency>

常量類

public class Constants {
    public static final String HOST = "47.98.109.138";
    public static final int PORT = 5672;
    public static final String USER_NAME = "study";
    public static final String PASSWORD = "study";
    public static final String VIRTUAL_HOST = "aaa";
    //通配符模式
    public static final String TOPIC_EXCHANGE = "topic.exchange";
    public static final String TOPIC_QUEUE1 = "topic_queue1";
    public static final String TOPIC_QUEUE2 = "topic_queue2";
}

編寫生產(chǎn)者代碼

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import rabbitmq.constant.Constants;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
 * 通配符模式生產(chǎn)者
 */
public class Producer {
    public static void main(String[] args) throws IOException, TimeoutException {
        //1. 建立連接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost(Constants.HOST);
        connectionFactory.setPort(Constants.PORT); //需要提前開放端口號
        connectionFactory.setUsername(Constants.USER_NAME);//賬號
        connectionFactory.setPassword(Constants.PASSWORD);  //密碼
        connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST); //虛擬主機
        Connection connection = connectionFactory.newConnection();
        //2. 開啟信道
        Channel channel = connection.createChannel();
        //3. 聲明交換機
        channel.exchangeDeclare(Constants.TOPIC_EXCHANGE, BuiltinExchangeType.TOPIC, true);
        //4. 聲明隊列
        channel.queueDeclare(Constants.TOPIC_QUEUE1, true, false, false, null);
        channel.queueDeclare(Constants.TOPIC_QUEUE2, true, false, false, null);
        //5. 綁定交換機和隊列
        channel.queueBind(Constants.TOPIC_QUEUE1, Constants.TOPIC_EXCHANGE, "*.a.*");
        channel.queueBind(Constants.TOPIC_QUEUE2, Constants.TOPIC_EXCHANGE, "*.*.b");
        channel.queueBind(Constants.TOPIC_QUEUE2, Constants.TOPIC_EXCHANGE, "c.#");
        //6. 發(fā)送消息
        String msg = "hello topic, my routingkey is ae.a.f....";
        channel.basicPublish(Constants.TOPIC_EXCHANGE,"ae.a.f", null, msg.getBytes());  //轉(zhuǎn)發(fā)到Q1
        String msg_b = "hello topic, my routingkey is ef.a.b....";
        channel.basicPublish(Constants.TOPIC_EXCHANGE,"ef.a.b", null, msg_b.getBytes()); //轉(zhuǎn)發(fā)到Q1和Q2
        String msg_c = "hello topic, my routingkey is c.ef.d....";
        channel.basicPublish(Constants.TOPIC_EXCHANGE,"c.ef.d", null, msg_c.getBytes());//轉(zhuǎn)發(fā)Q2
        System.out.println("消息發(fā)送成功");
        //7. 釋放資源
        channel.close();
        connection.close();
    }
}

編寫消費者1代碼

import com.rabbitmq.client.*;
import rabbitmq.constant.Constants;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class Consumer1 {
    public static void main(String[] args) throws IOException, TimeoutException {
        //1. 建立連接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost(Constants.HOST);
        connectionFactory.setPort(Constants.PORT); //需要提前開放端口號
        connectionFactory.setUsername(Constants.USER_NAME);//賬號
        connectionFactory.setPassword(Constants.PASSWORD);  //密碼
        connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST); //虛擬主機
        Connection connection = connectionFactory.newConnection();
        //2. 開啟信道
        Channel channel = connection.createChannel();
        //3. 聲明隊列
        channel.queueDeclare(Constants.TOPIC_QUEUE1,true,false,false,null);
        //4. 消費消息
        DefaultConsumer consumer = new DefaultConsumer(channel){
            //從隊列中收到消息, 就會執(zhí)行的方法
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("接收到消息:"+ new String(body));
            }
        };
        channel.basicConsume(Constants.TOPIC_QUEUE1, true, consumer);
    }
}

編寫消費者2代碼

import com.rabbitmq.client.*;
import rabbitmq.constant.Constants;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class Consumer2 {
    public static void main(String[] args) throws IOException, TimeoutException {
        //1. 建立連接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost(Constants.HOST);
        connectionFactory.setPort(Constants.PORT); //需要提前開放端口號
        connectionFactory.setUsername(Constants.USER_NAME);//賬號
        connectionFactory.setPassword(Constants.PASSWORD);  //密碼
        connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST); //虛擬主機
        Connection connection = connectionFactory.newConnection();
        //2. 開啟信道
        Channel channel = connection.createChannel();
        //3. 聲明隊列
        channel.queueDeclare(Constants.TOPIC_QUEUE2,true,false,false,null);
        //4. 消費消息
        DefaultConsumer consumer = new DefaultConsumer(channel){
            //從隊列中收到消息, 就會執(zhí)行的方法
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("接收到消息:"+ new String(body));
            }
        };
        channel.basicConsume(Constants.TOPIC_QUEUE2, true, consumer);
    }
}