快捷導(dǎo)航

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

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

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

通配符模式

概述

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

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

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

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

應(yīng)用場(chǎng)景

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

優(yōu)勢(shì)

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

代碼案例

引入依賴(lài)

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

常量類(lèi)

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";
}

編寫(xiě)生產(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); //需要提前開(kāi)放端口號(hào)
        connectionFactory.setUsername(Constants.USER_NAME);//賬號(hào)
        connectionFactory.setPassword(Constants.PASSWORD);  //密碼
        connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST); //虛擬主機(jī)
        Connection connection = connectionFactory.newConnection();
        //2. 開(kāi)啟信道
        Channel channel = connection.createChannel();
        //3. 聲明交換機(jī)
        channel.exchangeDeclare(Constants.TOPIC_EXCHANGE, BuiltinExchangeType.TOPIC, true);
        //4. 聲明隊(duì)列
        channel.queueDeclare(Constants.TOPIC_QUEUE1, true, false, false, null);
        channel.queueDeclare(Constants.TOPIC_QUEUE2, true, false, false, null);
        //5. 綁定交換機(jī)和隊(duì)列
        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();
    }
}

編寫(xiě)消費(fèi)者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); //需要提前開(kāi)放端口號(hào)
        connectionFactory.setUsername(Constants.USER_NAME);//賬號(hào)
        connectionFactory.setPassword(Constants.PASSWORD);  //密碼
        connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST); //虛擬主機(jī)
        Connection connection = connectionFactory.newConnection();
        //2. 開(kāi)啟信道
        Channel channel = connection.createChannel();
        //3. 聲明隊(duì)列
        channel.queueDeclare(Constants.TOPIC_QUEUE1,true,false,false,null);
        //4. 消費(fèi)消息
        DefaultConsumer consumer = new DefaultConsumer(channel){
            //從隊(duì)列中收到消息, 就會(huì)執(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);
    }
}

編寫(xiě)消費(fèi)者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); //需要提前開(kāi)放端口號(hào)
        connectionFactory.setUsername(Constants.USER_NAME);//賬號(hào)
        connectionFactory.setPassword(Constants.PASSWORD);  //密碼
        connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST); //虛擬主機(jī)
        Connection connection = connectionFactory.newConnection();
        //2. 開(kāi)啟信道
        Channel channel = connection.createChannel();
        //3. 聲明隊(duì)列
        channel.queueDeclare(Constants.TOPIC_QUEUE2,true,false,false,null);
        //4. 消費(fèi)消息
        DefaultConsumer consumer = new DefaultConsumer(channel){
            //從隊(duì)列中收到消息, 就會(huì)執(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);
    }
}