Spring 整合 RabbitMQ 實戰指南：Fanout 廣播與 Direct 路由模式完整配置

Spring 整合 RabbitMQ 實戰指南：Fanout 廣播與 Direct 路由模式完整配置

Spring AMQP 提供了 org.springframework.amqp.rabbit 包，聲明式配置 RabbitMQ 連接工廠、交換機、隊列、綁定與模板，極大簡化生產者/消費者開發。本文演示 Fanout 廣播模式與 Direct 路由模式的完整 XML 配置與代碼實現。

1. Maven 依賴

<dependency>
    <groupId>org.springframework.amqp</groupId>
    <artifactId>spring-rabbit</artifactId>
    <version>1.6.0.RELEASE</version>
</dependency>

2. 共享配置 amqp-share.xml

<rabbit:connection-factory id="connectionFactory"
    host="${rabbit.hosts}" port="${rabbit.port}"
    username="${rabbit.username}" password="${rabbit.password}"
    virtual-host="${rabbit.virtualHost}" channel-cache-size="50"/>

<rabbit:admin connection-factory="connectionFactory"/>

<!-- 隊列聲明：持久化、非自動刪除 -->
<rabbit:queue name="spittle.alert.queue.1" durable="true" auto-delete="false"/>
<rabbit:queue name="spittle.alert.queue.2" durable="true" auto-delete="false"/>
<rabbit:queue name="spittle.alert.queue.3" durable="true" auto-delete="false"/>

<!-- Fanout 交換機：廣播綁定三個隊列 -->
<rabbit:fanout-exchange id="spittle.fanout" name="spittle.fanout" durable="true">
    <rabbit:bindings>
        <rabbit:binding queue="spittle.alert.queue.1"/>
        <rabbit:binding queue="spittle.alert.queue.2"/>
        <rabbit:binding queue="spittle.alert.queue.3"/>
    </rabbit:bindings>
</rabbit:fanout-exchange>

3. 生產者配置與代碼

amqp-producer.xml

<import resource="amqp-share.xml"/>

<rabbit:template id="rabbitTemplate"
    connection-factory="connectionFactory"
    exchange="spittle.fanout"
    message-converter="jsonMessageConverter"/>

<bean id="jsonMessageConverter"
    class="org.springframework.amqp.support.converter.JsonMessageConverter"/>

生產者主程序

public class ProducerMain {
    public static void main(String[] args) throws Exception {
        ApplicationContext ctx = new ClassPathXmlApplicationContext("amqp/amqp-producer.xml");
        AmqpTemplate template = ctx.getBean("rabbitTemplate", AmqpTemplate.class);

        for (int i = 0; i < 20; i++) {
            Spittle spittle = new Spittle((long)i, null, "Hello " + i, new Date());
            template.convertAndSend(spittle); // 默認使用配置的 exchange
            System.out.println("Sent #" + i);
            Thread.sleep(5000);
        }
    }
}

Spittle 實體需實現 Serializable，字段含 id、message、postedTime 等。

4. 消費者配置與代碼

監聽器實現 MessageListener

public class SpittleAlertHandler implements MessageListener {
    @Override
    public void onMessage(Message message) {
        try {
            String body = new String(message.getBody(), "UTF-8");
            System.out.println("Received: " + body);
            // 可用 Jackson 反序列化為 Spittle 對象進一步處理
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
        }
    }
}

amqp-consumer.xml

<import resource="amqp-share.xml"/>

<rabbit:listener-container connection-factory="connectionFactory">
    <rabbit:listener ref="spittleListener" method="onMessage"
        queues="spittle.alert.queue.1,spittle.alert.queue.2,spittle.alert.queue.3"/>
</rabbit:listener-container>

<bean id="spittleListener" class="com.example.rabbitmq.SpittleAlertHandler"/>

消費者啟動

public class ConsumerMain {
    public static void main(String[] args) {
        new ClassPathXmlApplicationContext("amqp/amqp-consumer.xml");
        // 容器啟動即開始監聽
    }
}

5. Direct 路由模式擴展

交換機與綁定配置

<rabbit:queue id="spring-test-queue1" name="spring-test-queue1" durable="true" auto-delete="false"/>
<rabbit:queue name="spring-test-queue2" durable="true" auto-delete="false"/>

<rabbit:direct-exchange name="${rabbit.exchange.direct}" durable="true" id="${rabbit.exchange.direct}">
    <rabbit:bindings>
        <rabbit:binding queue="spring-test-queue1" key="spring.test.queueKey1"/>
        <rabbit:binding queue="spring-test-queue2" key="spring.test.queueKey2"/>
    </rabbit:bindings>
</rabbit:direct-exchange>

<rabbit:template exchange="${rabbit.exchange.direct}" id="rabbitTemplate" .../>

生產者指定 Routing Key

mqProducer.sendDateToQueue("spring.test.queueKey1", "Hello Key1");
mqProducer.sendDateToQueue("spring.test.queueKey2", "Hello Key2");

消費者分隊列監聽

<rabbit:listener-container connection-factory="connectionFactory" acknowledge="auto">
    <rabbit:listener queues="spring-test-queue1" method="onMessage" ref="queueListener"/>
    <rabbit:listener queues="spring-test-queue2" method="onMessage" ref="queueListener"/>
</rabbit:listener-container>

6. 關鍵點總結

• convertAndSend(Object) 默認發送到配置的 exchange，routing key 為空（Fanout 忽略 key）。
• Fanout：一條消息廣播給所有綁定隊列，適合事件通知、日志分發。
• Direct：根據 routing key 精確匹配隊列，適合業務分發、命令路由。
• JSON 序列化需實體類實現 Serializable，字段提供 getter/setter。
• 生產環境建議：手動 ACK（acknowledge="manual"）、死信隊列、TTL、發布確認、消費者限流（prefetchCount）。

Spring AMQP 的聲明式配置讓 RabbitMQ 整合變得極其簡潔，掌握交換機類型與綁定關係，即可靈活應對各種消息路由場景。