activemq實戰整合Spring之Topic模式

阿新 • • 發佈：2018-11-09

這一節我們看一下Topic模式下的訊息釋出是如何處理的。

applicationContext-ActiveMQ.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:amq="http://activemq.apache.org/schema/core"
        
xmlns:context="http://www.springframework.org/schema/context"
       xmlns:mvc="http://www.springframework.org/schema/mvc"
       xsi:schemaLocation="
http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-4.1.xsd
http://www.springframework.org/schema/context
http://www.springframework.org/schema/context/spring-context-4.1.xsd
http://www.springframework.org/schema/mvc
http://www.springframework.org/schema/mvc/spring-mvc-4.1.xsd
http://activemq.apache.org/schema/core
http://activemq.apache.org/schema/core/activemq-core-5.12.1.xsd" 
>

    <context:component-scan base-package="cn.edu.hust.activemq" />
    <mvc:annotation-driven />

    <amq:connectionFactory id="amqConnectionFactory"
                           brokerURL="tcp://127.0.0.1:61616"
                           userName="admin"
                           password 
="admin" />

    <!-- 配置JMS連線工廠 -->
    <bean id="connectionFactory"
          class="org.springframework.jms.connection.CachingConnectionFactory">
        <constructor-arg ref="amqConnectionFactory" />
        <property name="sessionCacheSize" value="100" />
    </bean>

    <!-- 定義訊息佇列（topic） -->
    <bean id="demoTopicDestination" class="org.apache.activemq.command.ActiveMQTopic">
        <!-- 設定訊息佇列的名字 -->
        <constructor-arg>
            <value>first-queue</value>
        </constructor-arg>
    </bean>

    <!-- 配置JMS模板（topic），Spring提供的JMS工具類，它傳送、接收訊息。 -->
    <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
        <property name="connectionFactory" ref="connectionFactory" />
        <property name="defaultDestination" ref="demoTopicDestination" />
        <property name="receiveTimeout" value="10000" />        
        <property name="pubSubDomain" value="true" />
    </bean>

    <!-- 配置訊息佇列監聽者（topic） -->
    <bean id="topicMessageListener" class="cn.edu.hust.activemq.filter.QueueMessageListener" />
    <bean id="topicMessageListener1" class="cn.edu.hust.activemq.filter.QueueMessageListener1" />

    <!-- 顯示注入訊息監聽容器（topic），配置連線工廠，監聽的目標是demoQueueDestination，監聽器是上面定義的監聽器 -->
    <bean id="queueListenerContainer"
          class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="connectionFactory" ref="connectionFactory" />
        <property name="destination" ref="demoTopicDestination" />
        <property name="messageListener" ref="topicMessageListener" />       
        <!--訊息接收超時 -->
        <property name="receiveTimeout" value="10000" />

    </bean>


    <bean id="queueListenerContainerB"
          class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="connectionFactory" ref="connectionFactory" />
        <property name="destination" ref="demoTopicDestination" />
        <property name="messageListener" ref="topicMessageListener1" />
        <!--訊息接收超時 -->
        <property name="receiveTimeout" value="10000" />
    </bean>  
</beans>
    
     1
     2
     3
     4
     5
     6
     7
     8
     9
     10
     11
     12
     13
     14
     15
     16
     17
     18
     19
     20
     21
     22
     23
     24
     25
     26
     27
     28
     29
     30
     31
     32
     33
     34
     35
     36
     37
     38
     39
     40
     41
     42
     43
     44
     45
     46
     47
     48
     49
     50
     51
     52
     53
     54
     55
     56
     57
     58
     59
     60
     61
     62
     63
     64
     65
     66
     67
     68
     69
     70
     71
     72

這裡與queue模式不一樣的地方在於訂閱者有一個或是多個，有幾個訂閱者就需要配置監聽器。

applicationContext.xml

<?xml version="1.0" encoding="UTF-8"?>
<!-- 查詢最新的schemaLocation 訪問 http://www.springframework.org/schema/ -->
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:context="http://www.springframework.org/schema/context"
       xmlns:mvc="http://www.springframework.org/schema/mvc"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xsi:schemaLocation="http://www.springframework.org/schema/aop
http://www.springframework.org/schema/aop/spring-aop-4.0.xsd
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/mvc
http://www.springframework.org/schema/mvc/spring-mvc-4.0.xsd">

    <!-- 指定Sping元件掃描的基本包路徑 -->
    <context:component-scan base-package="cn.edu.hust.activemq" >
        <!-- 這裡只掃描Controller，不可重複載入Service -->
        <context:include-filter type="annotation" expression="org.springframework.stereotype.Controller"/>
    </context:component-scan>
    <!-- 啟用MVC註解 -->
    <mvc:annotation-driven />

    <!-- JSP檢視解析器-->
    <bean class="org.springframework.web.servlet.view.InternalResourceViewResolver">
        <property name="prefix" value="/" />
        <property name="suffix" value=".jsp" />
        <!--  定義其解析檢視的order順序為1 -->
        <property name="order" value="1" />
    </bean>
</beans>
    
     1
     2
     3
     4
     5
     6
     7
     8
     9
     10
     11
     12
     13
     14
     15
     16
     17
     18
     19
     20
     21
     22
     23
     24
     25
     26
     27
     28
     29
     30
     31

web.xml

<?xml version="1.0" encoding="UTF-8"?>
<web-app xmlns="http://java.sun.com/xml/ns/javaee"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://java.sun.com/xml/ns/javaee
          http://java.sun.com/xml/ns/javaee/web-app_3_0.xsd"
         version="3.0">
<display-name>Archetype Created Web Application</display-name>
    <context-param>
        <param-name>contextConfigLocation</param-name>
        <param-value>classpath:applicationContext-ActiveMQ.xml</param-value>
    </context-param>

    <listener>
        <listener-class>org.springframework.web.context.ContextLoaderListener</listener-class>
    </listener>

    <servlet>
        <servlet-name>springMVC</servlet-name>
        <servlet-class>org.springframework.web.servlet.DispatcherServlet</servlet-class>
        <init-param>
            <param-name>contextConfigLocation</param-name>
            <param-value>classpath:applicationContext.xml</param-value>
        </init-param>
        <load-on-startup>1</load-on-startup>
    </servlet>
    <servlet-mapping>
        <servlet-name>springMVC</servlet-name>
        <url-pattern>/</url-pattern>
    </servlet-mapping>

    <!-- 處理編碼格式 -->
    <filter>
        <filter-name>characterEncodingFilter</filter-name>
        <filter-class>org.springframework.web.filter.CharacterEncodingFilter</filter-class>
        <init-param>
            <param-name>encoding</param-name>
            <param-value>UTF-8</param-value>
        </init-param>
        <init-param>
            <param-name>forceEncoding</param-name>
            <param-value>true</param-value>
        </init-param>
    </filter>
    <filter-mapping>
        <filter-name>characterEncodingFilter</filter-name>
        <url-pattern>/*</url-pattern>
    </filter-mapping>

</web-app>
    
     1
     2
     3
     4
     5
     6
     7
     8
     9
     10
     11
     12
     13
     14
     15
     16
     17
     18
     19
     20
     21
     22
     23
     24
     25
     26
     27
     28
     29
     30
     31
     32
     33
     34
     35
     36
     37
     38
     39
     40
     41
     42
     43
     44
     45
     46
     47
     48
     49

其餘部分與queue模式程式碼一樣，只是需要多配置一個監聽器，目錄結構如下：

這裡寫圖片描述

ProducerService.java

import javax.jms.Destination;


/**
 * Created by Administrator on 2017/5/3.
 */
public interface ProducerService {

    void sendMessage(Destination destination,final String msg);

    void sendMessage(final String msg);
}
    
     1
     2
     3
     4
     5
     6
     7
     8
     9
     10
     11
     12

ProducerServiceImpl.java

import cn.edu.hust.activemq.service.ProducerService;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;

/**
 * Created by Administrator on 2017/5/3.
 */
@Service
public class ProducerServiceImpl implements ProducerService {

    @Resource(name="jmsTemplate")
    private JmsTemplate jmsTemplate;

    @Override
    public void sendMessage(Destination destination, final String msg) {
        System.out.println(Thread.currentThread().getName()+" 向佇列"+destination.toString()+"傳送訊息--------->"+msg);
        jmsTemplate.send(destination, new MessageCreator() {
            public Message createMessage(Session session) throws JMSException {
                return session.createTextMessage(msg);
            }
        });
    }

    @Override
    public void sendMessage(final String msg) {
        String destination = jmsTemplate.getDefaultDestinationName();
        System.out.println(Thread.currentThread().getName()+" 向佇列"+destination+"傳送訊息-------->"+msg);
        jmsTemplate.send(new MessageCreator() {
            public Message createMessage(Session session) throws JMSException {
                return session.createTextMessage(msg);
            }
        });
    }
}
    
     1
     2
     3
     4
     5
     6
     7
     8
     9
     10
     11
     12
     13
     14
     15
     16
     17
     18
     19
     20
     21
     22
     23
     24
     25
     26
     27
     28
     29
     30
     31
     32
     33
     34
     35
     36
     37
     38
     39
     40
     41

ConsumerService.java

import javax.jms.Destination;
import javax.jms.TextMessage;
/**
 * Created by Administrator on 2017/5/3.
 */
public interface ConsumerService {

    TextMessage receive(Destination destination);
}
    
     1
     2
     3
     4
     5
     6
     7
     8
     9

ConsumerServiceImpl.java

import cn.edu.hust.activemq.service.ConsumerService;

import javax.jms.Destination;
import javax.jms.TextMessage;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import javax.jms.JMSException;

/**
 * Created by Administrator on 2017/5/3.
 */
@Service
public class ConsumerServiceImpl implements ConsumerService {
    @Resource(name="jmsTemplate")
    private JmsTemplate jmsTemplate;

    @Override
    public TextMessage receive(Destination destination){
        TextMessage textMessage = (TextMessage) jmsTemplate.receive(destination);
        try{
            System.out.println("從佇列" + destination.toString() + "收到了訊息：\t"
                    + textMessage.getText());
        } catch (JMSException e) {
            e.printStackTrace();
        }
        return textMessage;
    }
}
    
     1
     2
     3
     4
     5
     6
     7
     8
     9
     10
     11
     12
     13
     14
     15
     16
     17
     18
     19
     20
     21
     22
     23
     24
     25
     26
     27
     28
     29
     30

QueueMessageListener.java

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;

/**
 * Created by Administrator on 2017/5/3.
 */
public class QueueMessageListener implements MessageListener {
    public void onMessage(Message message) {
        TextMessage tm = (TextMessage) message;
        try {
            System.out.println("topicMessageListener監聽到了文字訊息：\t"
                    + tm.getText());
            //do something ...
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
}
    
     1
     2
     3
     4
     5
     6
     7
     8
     9
     10
     11
     12
     13
     14
     15
     16
     17
     18
     19
     20

QueueMessageListener1.java

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;

/**
 * Created by Administrator on 2017/5/3.
 */
public class QueueMessageListener1 implements MessageListener {
    public void onMessage(Message message) {
        TextMessage tm = (TextMessage) message;
        try {
            Syst

 
 
              
           
              
              
            
            相關推薦
			   
            
            
            
 

    

    
    activemq實戰整合Spring之Topic模式
       
  
  
  
   
   這一節我們看一下Topic模式下的訊息釋出是如何處理的。 
   applicationContext-ActiveMQ.xml配置： 
   <?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="htt 

  
 

    

    
    mybatis 整合spring之mapperLocations配置的問題
      any   引入   location   rop   myba   路徑   通過   user   configure    今天嘗試spring整合mybatis時遇到這麽一個問題，就是在配置sqlSessionFactory時是否要配置mapperLocations的問題。
 

<bean i 

  
 

    

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

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

  
 

    

    
    shiro學習筆記：整合spring之攔截器鏈執行流程
      
                一、環境準備搭建好spring + shiro整合環境（本文環境Spring 4.3.10.RELEASE + Shiro 1.4.0）後，編寫登入頁面如下：<html>
<head>
<title>登入頁</title>
&l 

  
 

    

    
    Java Web實戰04--Spring之屬性注入和構造器注入
      
								
								            
						
                

和前面一樣，建立maven工程，然後進行以下操作過程，
1、新建一個bag類，設定四個欄位，如下所示：
package com.yefeng.spring.spring2;


/**
 * @au 

  
 

    

    
    mybatis整合spring之mapperLocations和typeAliasesPackage(mapper-locations和type-aliases-package)
      
								
								            
						
                Spring整合<bean id="sessionFactory" class="org.mybatis.spring.SqlSessionFactoryBean">
    <pro 

  
 

    

    
    activemq實戰之整合Spring
       
  
  
  
   
   前面的理論準備已經很充分，這一節我們來實戰：將activemq整合到Spring框架才行中，因為Spring已經集成了JMS，這也為我們配置activermq帶來了方便。 
   1. Spring對jms的支援 
   因為Spring已經將JMS整合到框架裡面了，對jm 

  
 

    

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

import javax.annotation.Resour 

  
 

    

    
    SpringBoot2整合ActiveMQ實戰之釋出訂閱模式
       
 
 1、預設消費者並不會消費訂閱釋出型別的訊息，這是由於springboot預設採用的是p2p模式進行訊息的監聽 
      在配置檔案裡面，註釋掉 #spring.jms.pub-sub-domain=true 
   
 2.  //需要給topi 

  
 

    

    
    spring整合activemq傳送MQ訊息[Topic模式]例項
      <?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSc 

  
 

    

    
    利用Spring與ActiveMQ整合傳送、接收訊息例項(Queue與Topic模式)
      
                
利用Spring與ActiveMQ整合傳送、接收訊息例項，同時使用Queue與Topic兩種模式。
1.執行環境：Win10+Eclipse Java EE IDE Oxygen Release (4.7.0)+jdk1.8













3.向topic傳送訊 

  
 

    

    
    ActiveMQ實戰之 Topic釋出訂閱訊息
       
  
  
 分享一下我老師大神的人工智慧教程！零基礎，通俗易懂！http://blog.csdn.net/jiangjunshow
 
 也歡迎大家轉載本篇文章。分享知識，造福人民，實現我們中華民族偉大復興！
 
 
          

  
 

    

    
    Jetty實戰之 嵌入式Jetty整合Spring執行
                      本文連結：http://blog.csdn.net/kongxx/article/details/72271071. 首先修改pom.xml檔案，新增spring的依賴項<project xmlns="http://maven.apache.org/POM/4.0.0"  

  
 

    

    
    訊息佇列之非同步訊息基本概念以及ActiveMQ整合Spring常用用法介紹
       
 
 一 簡介 
 （1）非同步訊息： 
 所謂非同步訊息，跟RMI遠端呼叫、webservice呼叫是類似的，非同步訊息也是用於應用程式之間的通訊。但是它們之間的區別是： 
 
  RMI、Hession/Burlap、webservice等遠端呼叫機制是同步的。也就是說，當客戶端呼叫遠端方法時，客戶端 

  
 

    

    
    SpringBoot2整合ActiveMQ實戰之點對點訊息
       
 
 1.加入依賴         <!-- 整合訊息佇列ActiveMQ -->         <dependency>     &nbs 

  
 

    

    
    Java中介軟體JMS(四)之ActiveMQ整合spring之類轉換(三)
      
                
前幾章都是直接傳送MapMessage型別的資料,拿前面的例子來講,如果生產者傳送的是TextMessage，消費者也是必須TextMessage;如果我們自己要傳送的資料不是TextMessage型別,而消費者還是TextMessage的，那該怎麼辦？難道每次接受後都要增 

  
 

    

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

   <dependency>
            <groupId>org.springframework 

  
 

    

    
    Java中介軟體JMS(三)之ActiveMQ整合spring監聽器(二)
      
                
對於讓spring管理監聽的實現方式有兩種方法，一種是自己寫監聽器，然後交給spring的監聽介面卡管理，再由監聽容器管理監聽介面卡，另一種是寫一個實現MessageListener介面的類。第一種在第一章涉及到，但是沒有交給spring託管.其實實現的方法很簡單，在j2e 

  
 

    

    
    Java中介軟體JMS(四)之ActiveMQ整合spring之類轉換
      

前幾章都是直接傳送MapMessage型別的資料,拿前面的例子來講,如果生產者傳送的是TextMessage，消費者也是必須TextMessage;如果我們自己要傳送的資料不是TextMessage型別,而消費者還是TextMessage的，那該怎麼辦？難道每次接受後都要增加一個轉換方法麼？其實sprin 

  
 

    

    
    07-spring之三大框架的整合
      support   以及   RR   事務管理器   ati   body   png   方法   nat   1 三大框架整合理論

2 導包（42個）
1 hibernate
1 hibernate/lib/required

2 hibernate/lib/jpa | java pers