spring整合rabbitMq(基於topic和fanout模式)

阿新 • • 發佈：2019-01-23

本文程式碼樣例都是在spring整合環境下寫的，都測試通過。
pom檔案需要加入spring整合rabbitMq的依賴：

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

一、 rabbitMq的topic模式：

圖解：
這裡寫圖片描述

使用場景：傳送端不只按固定的routing key傳送訊息，而是按字串“匹配”傳送，接收端同樣如此。

傳送端spring的xml配置:

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xmlns:mvc="http://www.springframework.org/schema/mvc"
    xmlns:tx="http://www.springframework.org/schema/tx" 

    xmlns:jee="http://www.springframework.org/schema/jee"
    xmlns:p="http://www.springframework.org/schema/p"
    xmlns:aop="http://www.springframework.org/schema/aop"
    xmlns:context="http://www.springframework.org/schema/context"
    xmlns:task="http://www.springframework.org/schema/task"
    xmlns:rabbit="http://www.springframework.org/schema/rabbit" 

    xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.0.xsd
    http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-4.0.xsd
    http://www.springframework.org/schema/jee http://www.springframework.org/schema/jee/spring-jee-4.0.xsd
    http://www.springframework.org/schema/mvc http://www.springframework.org/schema/mvc/spring-mvc-4.0.xsd
    http://www.springframework.org/schema/tx  http://www.springframework.org/schema/tx/spring-tx-4.0.xsd
    http://www.springframework.org/schema/aop  http://www.springframework.org/schema/aop/spring-aop-4.0.xsd
    http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-4.0.xsd
    http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit-1.0.xsd">

    <!-- rabbitMQ配置 -->
    <bean id="rabbitConnectionFactory"
      class="org.springframework.amqp.rabbit.connection.CachingConnectionFactory">
        <constructor-arg value="192.168.1.11"/>
        <property name="username" value="root"/>
        <property name="password" value="lee13233"/>
        <property name="channelCacheSize" value="8"/>
        <property name="port" value="5672"></property>
    </bean>
    <rabbit:admin connection-factory="rabbitConnectionFactory"/>

    <!-- autoDelete:是否自動刪除 durable:持久化  -->
    <rabbit:queue name="test123queue" durable="true"/>
    <rabbit:queue name="test321queue" durable="true"/>

<!-- topic主題 -->
    <rabbit:topic-exchange name="leo.pay.topic.exchange" xmlns="http://www.springframework.org/schema/rabbit" durable="true">
      <bindings>
        <binding queue="test123queue" pattern="*.*.test123" />
        <binding queue="test321queue" pattern="test321.#" />
      </bindings>
    </rabbit:topic-exchange>

    <!-- 建立rabbitTemplate 訊息模板類 -->
    <bean id="rabbitTemplate" class="org.springframework.amqp.rabbit.core.RabbitTemplate">
        <constructor-arg ref="rabbitConnectionFactory"></constructor-arg>
    </bean>

</beans>

傳送端的java程式碼：

@Test
    public void testRabbitMq() throws Exception {
        RabbitTemplate rabbitTemplate = (RabbitTemplate) LeoContext.getContext().getApplication().getBean("rabbitTemplate");
        //第二個引數為路由key(routingKey)的值，當路由可以為test321.hello.test123時，兩個消費佇列都可以收到訊息，當值為test321.hello.aaa時，只有綁定了test321.#的佇列才可以收到訊息，當值為ta1.hello.test123，只有綁定了*.*.test123的佇列才可收到訊息
        for(int i = 1; i <= 10; i++) {
            String str = "hello" + i;
        rabbitTemplate.send("leo.pay.topic.exchange", "test321.hello.test123", new Message(str.getBytes(), new MessageProperties()));
        }
    }

接收端(我配置的接收端與傳送端不在同一個專案)spring的xml配置：

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xmlns:mvc="http://www.springframework.org/schema/mvc"
    xmlns:tx="http://www.springframework.org/schema/tx"
    xmlns:jee="http://www.springframework.org/schema/jee"
    xmlns:p="http://www.springframework.org/schema/p"
    xmlns:aop="http://www.springframework.org/schema/aop"
    xmlns:context="http://www.springframework.org/schema/context"
    xmlns:task="http://www.springframework.org/schema/task"
    xmlns:rabbit="http://www.springframework.org/schema/rabbit"
    xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.0.xsd
    http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-4.0.xsd
    http://www.springframework.org/schema/jee http://www.springframework.org/schema/jee/spring-jee-4.0.xsd
    http://www.springframework.org/schema/mvc http://www.springframework.org/schema/mvc/spring-mvc-4.0.xsd
    http://www.springframework.org/schema/tx  http://www.springframework.org/schema/tx/spring-tx-4.0.xsd
    http://www.springframework.org/schema/aop  http://www.springframework.org/schema/aop/spring-aop-4.0.xsd
    http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-4.0.xsd
    http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit-1.0.xsd"
    default-autowire="byName">

    <!-- rabbitMQ配置 -->
    <bean id="rabbitConnectionFactory"
      class="org.springframework.amqp.rabbit.connection.CachingConnectionFactory">
        <constructor-arg value="192.168.1.11"/>
        <property name="username" value="leo"/>
        <property name="password" value="lee31211"/>
        <property name="channelCacheSize" value="8"/>
        <property name="port" value="5672"></property>
    </bean>
    <rabbit:admin connection-factory="rabbitConnectionFactory"/>

    <rabbit:queue name="test123queue" durable="true" />
    <rabbit:queue name="test321queue" durable="true" />

    <!-- 該處是指將路由leo.pay.topic.exchange與兩個佇列繫結在一塊，也可以在rabbitMq的控制檯上手動繫結，手動繫結之後，該處程式碼可以省略，其實發送端已經繫結過了，也沒必要綁定了，所以該程式碼可以省略 -->
 <!--   <rabbit:topic-exchange name="leo.pay.topic.exchange" xmlns="http://www.springframework.org/schema/rabbit" durable="true"> -->
<!--      <bindings> -->
<!--        <binding queue="test123queue" pattern="test123.*" /> -->
<!--        <binding queue="test321queue" pattern="test321.*" /> -->
<!--      </bindings> -->
<!--    </rabbit:topic-exchange> -->

    <!-- 啟動兩個佇列對應的監聽(消費者) -->
    <bean id="detailQueueConsumer" class="com.leo.website.cousumer.DetailQueueConsumer"></bean>
    <bean id="testQueueConsumer" class="com.leo.website.cousumer.TestQueueConsumer"></bean>

    <!-- 將兩個佇列加入監聽容器中，每個佇列的監聽都對應一個監聽器 -->
    <rabbit:listener-container connection-factory="rabbitConnectionFactory" concurrency= "8">
        <rabbit:listener queues="test123queue" ref="detailQueueConsumer" method="onMessage"/>
        <rabbit:listener queues="test321queue" ref="testQueueConsumer" method="onMessage"/>
    </rabbit:listener-container>

</beans>

接收端java程式碼(只列出一個監聽，另外一個類似)：

public class DetailQueueConsumer implements MessageListener {
    @Override
    public void onMessage(Message message) {
        System.out.println("DetailQueueConsumer: " + new String(message.getBody()));
    }
}

二、 rabbitMq的fanout模式(釋出者訂閱者模式)：

圖解：
這裡寫圖片描述

使用場景：釋出、訂閱模式，傳送端傳送廣播訊息，多個接收端接收。

傳送端spring的xml配置：

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xmlns:mvc="http://www.springframework.org/schema/mvc"
    xmlns:tx="http://www.springframework.org/schema/tx"
    xmlns:jee="http://www.springframework.org/schema/jee"
    xmlns:p="http://www.springframework.org/schema/p"
    xmlns:aop="http://www.springframework.org/schema/aop"
    xmlns:context="http://www.springframework.org/schema/context"
    xmlns:task="http://www.springframework.org/schema/task"
    xmlns:rabbit="http://www.springframework.org/schema/rabbit"
    xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.0.xsd
    http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-4.0.xsd
    http://www.springframework.org/schema/jee http://www.springframework.org/schema/jee/spring-jee-4.0.xsd
    http://www.springframework.org/schema/mvc http://www.springframework.org/schema/mvc/spring-mvc-4.0.xsd
    http://www.springframework.org/schema/tx  http://www.springframework.org/schema/tx/spring-tx-4.0.xsd
    http://www.springframework.org/schema/aop  http://www.springframework.org/schema/aop/spring-aop-4.0.xsd
    http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-4.0.xsd
    http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit-1.0.xsd">

    <!-- rabbitMQ配置 -->
    <bean id="rabbitConnectionFactory"
      class="org.springframework.amqp.rabbit.connection.CachingConnectionFactory">
        <constructor-arg value="192.168.1.12"/>
        <property name="username" value="root"/>
        <property name="password" value="lee323"/>
        <property name="channelCacheSize" value="8"/>
        <property name="port" value="5672"></property>
    </bean>
    <rabbit:admin connection-factory="rabbitConnectionFactory"/>

    <rabbit:queue name="test123queue" durable="true"/>
    <rabbit:queue name="test321queue" durable="true"/>

    <rabbit:fanout-exchange name="leo.pay.fanout.exchange" xmlns="http://www.springframework.org/schema/rabbit" durable="true"> 
    </rabbit:fanout-exchange>

    <!-- 建立rabbitTemplate 訊息模板類 -->
    <bean id="rabbitTemplate" class="org.springframework.amqp.rabbit.core.RabbitTemplate">
        <constructor-arg ref="rabbitConnectionFactory"></constructor-arg>
    </bean>

</beans>

傳送端java程式碼：

@Test
    public void testRabbitMq() throws Exception {
        RabbitTemplate rabbitTemplate = (RabbitTemplate) LeoContext.getContext().getApplication().getBean("rabbitTemplate");
        //往名字為leo.pay.fanout.exchange的路由裡面傳送資料，客戶端中只要是與該路由繫結在一起的佇列都會收到相關訊息，這類似全頻廣播，傳送端不管佇列是誰，都由客戶端自己去繫結，誰需要資料誰去繫結自己的處理佇列。
        for(int i = 1; i <= 10; i++) {
            String str = "hello" + i;
rabbitTemplate.send("leo.pay.fanout.exchange", "", new Message(str.getBytes(), new MessageProperties()))
        }
    }

客戶端spring的xml配置(我舉例的配置客戶端與傳送端不在同一專案下)：

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xmlns:mvc="http://www.springframework.org/schema/mvc"
    xmlns:tx="http://www.springframework.org/schema/tx"
    xmlns:jee="http://www.springframework.org/schema/jee"
    xmlns:p="http://www.springframework.org/schema/p"
    xmlns:aop="http://www.springframework.org/schema/aop"
    xmlns:context="http://www.springframework.org/schema/context"
    xmlns:task="http://www.springframework.org/schema/task"
    xmlns:rabbit="http://www.springframework.org/schema/rabbit"
    xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.0.xsd
    http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-4.0.xsd
    http://www.springframework.org/schema/jee http://www.springframework.org/schema/jee/spring-jee-4.0.xsd
    http://www.springframework.org/schema/mvc http://www.springframework.org/schema/mvc/spring-mvc-4.0.xsd
    http://www.springframework.org/schema/tx  http://www.springframework.org/schema/tx/spring-tx-4.0.xsd
    http://www.springframework.org/schema/aop  http://www.springframework.org/schema/aop/spring-aop-4.0.xsd
    http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-4.0.xsd
    http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit-1.0.xsd"
    default-autowire="byName">

    <!-- rabbitMQ配置 -->
    <bean id="rabbitConnectionFactory"
      class="org.springframework.amqp.rabbit.connection.CachingConnectionFactory">
        <constructor-arg value="192.168.1.13"/>
        <property name="username" value="root"/>
        <property name="password" value="lee2342"/>
        <property name="channelCacheSize" value="8"/>
        <property name="port" value="5672"></property>
    </bean>
    <rabbit:admin connection-factory="rabbitConnectionFactory"/>

    <rabbit:queue name="test123queue" durable="true" />
    <rabbit:queue name="test321queue" durable="true" />

    <!-- 該處把需要資料的佇列與路由繫結一起，如果手動在控制檯繫結就不需要此程式碼 -->
    <rabbit:fanout-exchange name="leo.pay.fanout.exchange" xmlns="http://www.springframework.org/schema/rabbit" durable="true">
        <rabbit:bindings>
            <rabbit:binding queue="test123queue"></rabbit:binding>
            <rabbit:binding queue="test321queue"></rabbit:binding> 
        </rabbit:bindings> 
    </rabbit:fanout-exchange> 


    <bean id="detailQueueConsumer" class="com.leo.website.cousumer.DetailQueueConsumer"></bean>
    <bean id="testQueueConsumer" class="com.leo.website.cousumer.TestQueueConsumer"></bean>

    <rabbit:listener-container connection-factory="rabbitConnectionFactory" concurrency= "8">
        <rabbit:listener queues="test123queue" ref="detailQueueConsumer" method="onMessage"/>
        <rabbit:listener queues="test321queue" ref="testQueueConsumer" method="onMessage"/>
    </rabbit:listener-container>

</beans>

spring整合rabbitMq(基於topic和fanout模式)

本文程式碼樣例都是在spring整合環境下寫的，都測試通過。 pom檔案需要加入spring整合rabbitMq的依賴： <dependency> <groupId>org.springframework

springboot+RabbitMq之topic和fanout

1.建立一個springboot專案，匯入如下maven座標 <dependency> <groupId>org.springframework.boot</groupId> <artifactId>

Spring-boot 整合RabbitMQ ------基於註解的開發

spring: rabbitmq: host: 192.168.3.193 username: guest password: guest package com.shi.amqp; import org.springframework.amqp.rab

SPRING 整合 activemq 的 topic 模式

概要 activemq 支援兩種模式: 1.佇列模式 2. 釋出訂閱者模式，topic有一個主題可以有多個訂閱者。這種情況可以將一個訊息，分發到多個消費者。實現程式碼 1.生產者 import java.util.Map; import javax.annotation.Resour

struts2和spring整合，基於註解的方法

直接碼程式碼專案目錄1.匯入struts和spring各自需要的jar包，再匯入兩者整合需要的jar包2.UserDao.javapackage com.ssh.dao; import org.springframework.stereotype.Repository;

ActiveMQ5.0實戰三:使用Spring發送,消費topic和queue消息

cme 擁有支持 add gin logs ges .get sum 實戰一 , 實戰二介紹了ActiveMQ的基本概念和配置方式. 本篇將通過一個實例介紹使用spring發送,消費topic, queue類型消息的方法. 不懂topic和queue的google 之

Django之基於session和CBV模式裝飾器實現使用者認證

models.py from django.db import models # Create your models here. class User(models.Model): name = models.CharField(max_length=32) pwd = mode

Spring整合RabbitMQ帶測試用例Demo完整程式碼

專案結構 jar包  <dependency> <groupId>com.rabbitmq</groupId> <artifactId&

訊息佇列-----生成者 Spring整合rabbitmq

最近做Spring結合Rabbitmq，然後往佇列裡面傳送了一個訊息，如何檢視訊息開啟:http://127.0.0.1:15672 輸入使用者名稱和密碼:找到你的佇列我們可以找到自己的佇列，並且發現裡面有一個訊息，我們點選這個佇列，進入: 可以看到這個佇列的這些基本

（七）RabbitMQ訊息佇列-通過fanout模式將訊息推送到多個Queue中

前面第六章我們使用的是direct直連模式來進行訊息投遞和分發。本章將介紹如何使用fanout模式將訊息推送到多個佇列。有時我們會遇到這樣的情況，多個功能模組都希望得到完整的訊息資料。例如一個log的訊息，一個我們希望輸出在螢幕上實時監控，另外一個使用者持久化日誌。這時就可以使用fano

SpringBoot整合RabbitMQ之Topic Exchange萬用字元交換機(學習總結)

一、簡介 Topic Exchange交換機也叫萬用字元交換機，我們在傳送訊息到Topic Exchange的時候不能隨意指定route key（應該是由一系列點號連線的字串，一般會與binding key有關聯，route key的長度一般不能超過255個位元組)。同理，

RabbitMQ的學習(六)：Spring整合RabbitMQ並進行兩個不同的系統間的呼叫

上一篇文章已經講解了spring整合RabbitMQ的例子，目前所有的例子都是簡單明瞭，程式設計師需要有舉一反三的能力，所以若覺得例子太簡單，請手下留情。。這篇文章將弄兩個系統，然後進行系統間的通訊，通過該例子，希望你能對分散式，叢集間的通訊有所瞭解。新建一個能啟動的web專

RabbitMQ的學習(五)：（最全面）Spring整合RabbitMQ進行RPC遠端呼叫

前面用java簡單對rabbitmq進行了學習，但rabbitmq的功能遠非如此，配合Spring或Springboot進行開發，才是java工程師做的工作，下面直接進行實戰：一、建立SSM專案 PS初期web專案下載地址(單純的web，用於練習)： https://downloa

(三) RabbitMQ實戰教程(面向Java開發人員)之Spring整合RabbitMQ

Spring整合RabbitMQ Spring AMQP是對AMQP協議的抽象和封裝，從官方網站上得知它是由兩個專案組成的(spring-amqp和spring-rabbit)。在使用Spring整合RabbitMQ時我們主要關注三個核心介面(MessageL

Spring整合Rabbitmq收集Logback日誌，利用進行Logstash資料整理儲存到Elasticsearch中

專案中我們常用的是把Logback列印的日誌儲存到檔案中儲存到硬碟上，這樣不利於日誌的收集和分析。以下演示在SpringBoot中通過rabbitmq收集logback日誌儲存到Elasticsearch中。環境準備：安裝RabbitMQ，安裝Elasticsearc

【RabbitMQ】spring整合rabbitmq 檔案配置、demo完整示例

之前講了基於java配置的簡單的小demo，現在結合一下spring進行整合。其實原理跟之前差不多的，只是改成了xml的配置。1、建立工廠連線<rabbit:connection-factory id="connectionFactory"

Spring整合RabbitMQ

簡介為了更好更快速的去開發Rabbitmq應用,spring封裝client ;目的是簡化我們的使用. 開發整合 maven依賴 org.s

spring整合rabbitmq實現訊息佇列

RabbitTemplate概念瞭解 RabbitMQ是由Erlang(愛立信公司)語言開發，實現Advanced Message Queuing Protocol (AMQP高階訊息佇列協議)的訊息中介軟體。訊息中介軟體主要用於元件之間的解耦，訊息的傳送者無需知道

Spring 整合 RabbitMQ

pom.xml <dependency> <groupId>org.springframework.amqp</groupId> <artifactId>spring-rabb

RabbitMQ （九） Spring整合RabbitMQ(1)

前面幾篇講解了如何使用rabbitMq，這一篇主要講解spring整合rabbitmq。首先引入配置檔案org.springframework.amqp，如下 <dependency> <groupId>com.rabbitmq&

spring整合rabbitMq(基於topic和fanout模式)

一、 rabbitMq的topic模式：

二、 rabbitMq的fanout模式(釋出者訂閱者模式)：

相關推薦