JMS+activeMq+spring簡單學習及模擬處理新聞日誌場景

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

在介紹JMS的知識之前，先說說我瞭解學習JMS的原因。

公司有一個新聞專案，我們知道新聞網站都是面向所有使用者群的，它不但要求新聞的實時性，還要求網頁的響應時間必須很快，

你想啊，如果你在某個網站看新聞，如果網頁響應比較慢，那誰願意來你網站看新聞啊，所以新聞網站的網頁都進行靜態化、新聞

頁所需資料都放在快取系統(比如memcache之類)等等一些處理，也就是說使用者請求訪問某個頁面時，伺服器要儘可能快響應，這就

要求我們後臺，不能像一些企業應用系統（比如人事管理系統之類）那樣，每個請求所需的資料都從資料庫查詢，並且不止查詢

一次。因為這些應用系統都是有特定的使用人群的，併發量不大，所以哪怕不停查詢資料庫，響應也夠快，使用者也能接受。新聞

這專案，現在有個需求，需要收集記錄使用者的訪問日誌，比如使用者訪問了哪些新聞頻道列表，訪問了哪些新聞詳情等等，為統計

系統提供資料。這樣問題來了，網頁請求，連資料庫查詢都儘量避免，更別說執行日誌寫操作，要知道往資料庫裡面插資料，如

果表的資料量很大了，並且又有索引，那插入一條新資料所需的時間就很大的，如果併發量高時，還有可能阻塞，這就嚴重影響

網頁請求的響應速度了。有些小夥伴就說了，那我新啟動一個執行緒單獨對日誌寫不就行了嗎？對，這是一種方法，但是我們要注

意，如果請求非常多，那新建立的執行緒就會非常多，會嚴重影響伺服器的效能的。看了下

專案程式碼，專案中採用的是利用JMS來做。

一看上面的圖，這不是生產者和消費者模式嗎?是的，沒錯。看上面圖，可以看出，新聞應用伺服器、mq伺服器、日誌處理

伺服器都是分開的，新聞應用伺服器只需要把日誌資料傳送到mq伺服器就行了，至於日誌資料是怎樣處理的，插入資料庫的速度

如何，是否成功等，新聞應用伺服器不必關心，這樣一來就能儘可能快的提高使用者的請求響應了。

好了，接下來，我們學習瞭解一下關於JMS的一些基礎知識。

下載安裝執行activeMq

網路上下載一個activeMq，然後執行它

可以看到，預設的連線地址和埠是tcp://127.0.0.1:61616,同時還可以通過瀏覽器訪問http://localhost:8161/來檢視activeMq裡

面佇列等一些基礎資料資訊。現在瀏覽器開啟http://localhost:8161/

我們選擇Queue Views下的XML來看下

可以看到你佇列名稱，以及一些狀態資料

JMS基本概念

JMS(Java Message Service) 即Java訊息服務。它提供標準的產生、傳送、接收訊息的介面簡化企業應用的開發。它支援兩種訊息通訊模型：

點到點（point-to-point）（P2P）模型和釋出/訂閱（Pub/Sub）模型。P2P 模型規定了一個訊息只能有一個接收者;Pub/Sub 模型允許一個

訊息可以有多個接收者。
對於點到點模型，訊息生產者產生一個訊息後，把這個訊息傳送到一個Queue（佇列）中，然後訊息接收者再從這個Queue中讀取，一旦

這個訊息被一個接收者讀取之後，它就在這個Queue中消失了，所以一個訊息只能被一個接收者消費。
與點到點模型不同，釋出/訂閱模型中，訊息生產者產生一個訊息後，把這個訊息傳送到一個Topic中，這個Topic可以同時有多個接收者

在監聽，當一個訊息到達這個Topic之後，所有訊息接收者都會收到這個訊息。

簡單的講，點到點模型和釋出/訂閱模型的區別就是前者是一對一，後者是一對多。

Queue的簡單例子1

現在我們來看個簡單例子，用來模擬上面的新聞應用伺服器、activeMq、日誌處理伺服器這三方過程.首先啟

動activeMq服務，然後建立一個專案jms_produce用來生產資料，jms_consumer用來消費資料

其中jms_produce的MessageSend.java的程式碼如下

package com.test;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Message;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQQueue;
public class MessageSend {
    public static void main(String[] args) throws Exception {
    	//建立連線工廠
        ConnectionFactory factory = new ActiveMQConnectionFactory("tcp://127.0.0.1:61616");
        //獲取連線物件
        Connection connection = factory.createConnection();
        //開始連線
        connection.start();
       //建立佇列，並且佇列名稱取名為"testQueue"
        Queue queue = new ActiveMQQueue("testQueue");
       //建立回話，具體的引數就不詳細介紹說明了，可以看API
        final Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        //建立訊息生產者
        MessageProducer producer = session.createProducer(queue);
       //向activeMq伺服器傳送7個數據，都位於佇列testQueue中
        for(int i=0;i<7;i++){
        	 Message message = session.createTextMessage("Hello JMS "+i);//如果是物件傳送，可用createObjectMessage(...)
             producer.send(message);
        }
        //關閉連結
        connection.close();
    }
}

jms_consumer中的MessageRecieve.java程式碼如下

package com.test;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQQueue;
public class MessageReceive {
    public static void main(String[] args) throws Exception {
    	//建立連線工廠
        ConnectionFactory factory = new ActiveMQConnectionFactory("tcp://127.0.0.1:61616");
        //獲取連線物件
        Connection connection = factory.createConnection();
        //開始連線
        connection.start();
       
      //建立佇列，並且佇列名稱取名為"testQueue"
        Queue queue = new ActiveMQQueue("testQueue");
       //建立回話，具體的引數就不詳細介紹說明了，可以看API
        final Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
       //建立訊息消費者，這個消費只能消費testQueue佇列中的資料
        MessageConsumer comsumer = session.createConsumer(queue);
        while(true){
        	Message recvMessage = comsumer.receive();//這裡會進行阻塞
        	System.out.println(((TextMessage)recvMessage).getText());
        }
    }
}

可以看到生產者往佇列testQueue中生產了7個數據。現在我們執行jms_produce，消費專案jms_consumer暫時不要啟動，因為我們要看看，生產者往activeMq中發數

據時，testQueue的狀態資料，然後啟動jms_consumer消費這消費後,activeMq中的testQueue的狀況。好了，現在我們執行jms_produce，然後瀏覽器中看到，

下面資訊（不記得怎麼用瀏覽器看activeMq中的資訊的，請回頭看看上面）

然後，我們再執行jms_consumer專案，把佇列testQueue中的資料給消費掉

控制檯

size有多少個生產者正在生產，應為上面jms_produce生產了7個數據後，它就直接關閉了，所以

這裡看到size=0，consumerCount表示正在有多少個消費者在連線這個佇列，看上面程式碼我們知道jms_consumer

一直在等待，一直在連線activeMq，所以consumerCount=1,enqueueCount表示佇列中有多少個數據，

dequueueCount表示有多少個數據被消費過。

Queue的簡單例子2

訊息的消費者接收訊息可以採用兩種方式：

1、consumer.receive() 或 consumer.receive(int timeout)；
2、註冊一個MessageListener。

採用第一種方式，訊息的接收者會一直等待下去，直到有訊息到達，或者超時。後一種方式會註冊一個監聽器，當有訊息到達的時候，

會回撥它的onMessage()方法。下面舉例說明：

我們把上面的jms_consumer總的MessageRecive.java稍微修改下

package com.test;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Message;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQQueue;
public class MessageSend {
    public static void main(String[] args) throws Exception {
    	//建立連線工廠
        ConnectionFactory factory = new ActiveMQConnectionFactory("tcp://127.0.0.1:61616");
        //獲取連線物件
        Connection connection = factory.createConnection();
        //開始連線
        connection.start();
       //建立佇列，並且佇列名稱取名為"testQueue"
        Queue queue = new ActiveMQQueue("testQueue");
       //建立回話，具體的引數就不詳細介紹說明了，可以看API
        final Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        //建立訊息生產者
        MessageProducer producer = session.createProducer(queue);
       //向activeMq伺服器傳送7個數據，都位於佇列testQueue中
        for(int i=0;i<7;i++){
        	 Message message = session.createTextMessage("Hello JMS "+i);//如果是物件傳送，可用createObjectMessage(...)
             producer.send(message);
        }
        //關閉連結
        connection.close();
    }
}

能達到同樣的效果。

上面都是一個生產者，一個消費者，下面我們來看看一個生產者，多個消費者，消費結果是怎樣的。

對jms_consumer中的MessageRecive.java修改下

package com.test;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQQueue;
public class MessageReceive {
    public static void main(String[] args) throws Exception {
    	//建立連線工廠
        ConnectionFactory factory = new ActiveMQConnectionFactory("tcp://127.0.0.1:61616");
        //獲取連線物件
        Connection connection = factory.createConnection();
        //開始連線
        connection.start();
       
      //建立佇列，並且佇列名稱取名為"testQueue"
        Queue queue = new ActiveMQQueue("testQueue");
       //建立回話，具體的引數就不詳細介紹說明了，可以看API
        final Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
       //建立訊息消費者，這個消費只能消費testQueue佇列中的資料
       //消費者1
        MessageConsumer comsumer1 = session.createConsumer(queue);
        //註冊消費監聽器
        comsumer1.setMessageListener(new MessageListener(){
                   @Override
                   public void onMessage(Message m) {
                       TextMessage textMsg = (TextMessage) m;
                       try {
                           System.out.println("消費者1"+textMsg.getText());
                       } catch (JMSException e) {
                           e.printStackTrace();
                       }
                   }
                  
               });
        
      //消費者2
        MessageConsumer comsumer2 = session.createConsumer(queue);
        //註冊消費監聽器
        comsumer2.setMessageListener(new MessageListener(){
                   @Override
                   public void onMessage(Message m) {
                       TextMessage textMsg = (TextMessage) m;
                       try {
                           System.out.println("消費者2:"+textMsg.getText());
                       } catch (JMSException e) {
                           e.printStackTrace();
                       }
                   }
                  
               });
    }
}

先啟動消費者jms_consumer,然後再啟動生產者，可以看到消費者的列印結果

發現，一個數據只能被消費一次，並且無法確定哪個資料被哪個消費者消費。

Topic的簡單例子

與Queue不同的是，Topic實現的是釋出/訂閱模型，也就是一個數據可以被多個消費者消費。

將上面生產者和消費者中程式碼的

Queue queue = new ActiveMQQueue("testQueue");

修改為

Topic topic= new ActiveMQTopic("testTopic");

先啟動消費者，再執行生產者，看到列印結果如下，每個資料都被所有消費者消費

JMS的一些深入知識

一個訊息物件分為三部分：訊息頭(Headers)，屬性（Properties）和訊息體（Payload）。對於StreamMessage和MapMessage，

訊息本身就有特定的結構，而對於TextMessage，ObjectMessage和BytesMessage是無結構的。一個訊息可以包含一

些重要的資料或者僅僅是一個事件的通知。

訊息的Headers部分通常包含一些訊息的描述資訊，它們都是標準的描述資訊。包含下面一些值：

　　JMSDestination
訊息的目的地，Topic或者是Queue。

　　JMSDeliveryMode
訊息的傳送模式：persistent或nonpersistent。前者表示訊息在被消費之前，如果JMS提供者DOWN了，重新啟動後消

息仍然存在。後者在這種情況下表示訊息會被丟失。可以通過下面的方式設定：
Producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

       JMSTimestamp
       當呼叫send()方法的時候，JMSTimestamp會被自動設定為當前事件。可以通過下面方式得到這個值：
       long timestamp = message.getJMSTimestamp();

　　JMSExpiration
表示一個訊息的有效期。只有在這個有效期內，訊息消費者才可以消費這個訊息。預設值為0，表示訊息永不過期。

可以通過下面的方式設定：
       producer.setTimeToLive(3600000); //有效期1小時（1000毫秒 * 60秒 * 60分）

　　JMSPriority
       訊息的優先順序。0-4為正常的優先順序，5-9為高優先順序。可以通過下面方式設定：
       producer.setPriority(9);

　　JMSMessageID
       一個字串用來唯一標示一個訊息。

　　JMSReplyTo
       有時訊息生產者希望消費者回復一個訊息，JMSReplyTo為一個Destination，表示需要回復的目的地。當然消費者可以不理會它。

　　JMSCorrelationID
       通常用來關聯多個Message。例如需要回復一個訊息，可以把JMSCorrelationID設定為所收到的訊息的JMSMessageID。

　　JMSType
       表示訊息體的結構，和JMS提供者有關。

　　JMSRedelivered
       如果這個值為true，表示訊息是被重新發送了。因為有時消費者沒有確認他已經收到訊息或者JMS

提供者不確定消費者是否已經收到。

除了Header，訊息傳送者可以新增一些屬性(Properties)。這些屬性可以是應用自定義的屬性，JMS定義的屬性

和JMS提供者定義的屬性。我們通常只適用自定義的屬性。

JMS+ActiveMq+Spring整合模擬處理新聞日誌場景

新建專案jms_spring_produce模擬新聞專案,專案jms_spring_consumer模擬日誌處理專案,

為了便捷，專案都是用maven結構搭建的。

下面會貼部分原始碼，完整的專案請下載

生產者和消費者之間的資料互動，採用了自定義的類。由於生產者和消費者位於兩個獨立

的專案中，所以這些日誌物件（如截圖上的LoginLogDto.java/NewsVisitLogDto.java）都必

須序列號，並且生產者和消費之間這些物件都必須一模一樣。如果覺得麻煩，那可以通過

json來進行互動，或者將這些日誌物件抽出來作為一個獨立的jar專案，maven管理很方便的。

由於程式碼較多，又有配置之類的，這裡就不一一全部貼出來了，就貼部分.

LoginLogDto.java

package com.fei.dto;

import java.io.Serializable;
import java.util.Date;
/**
 * 模擬使用者登入日誌
 * @author weijianfei
 *
 */
public class LoginLogDto implements Serializable{

	
	/**
	 * 
	 */
	private static final long serialVersionUID = 1L;
	
	private String userName;
	private Date loginTime;
	private String ip;
	
	public String getUserName() {
		return userName;
	}
	public void setUserName(String userName) {
		this.userName = userName;
	}
	public Date getLoginTime() {
		return loginTime;
	}
	public void setLoginTime(Date loginTime) {
		this.loginTime = loginTime;
	}
	public String getIp() {
		return ip;
	}
	public void setIp(String ip) {
		this.ip = ip;
	}
	
}

UserLoginLogProducer.java

package com.fei.service.jmsproducer;

import java.util.List;




import javax.jms.Destination;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.stereotype.Service;

import com.fei.dto.LoginLogDto;

@Service
public class UserLoginLogProducer {
	private static Logger logger = LoggerFactory.getLogger(UserLoginLogProducer.class);
	/**
	 * JMS模板
	 */
	@Autowired
	private JmsTemplate jmsTemplate;
	/**
	 * 使用者行為佇列
	 */
	@Autowired
	@Qualifier("userLoginQueue")
	private Destination userLoginQueue;

	
	public void sendMessage(LoginLogDto loginLog) {
		try{
			jmsTemplate.convertAndSend(this.userLoginQueue, loginLog);
		}catch(Exception e){
			logger.error("傳送使用者登入記錄訊息出錯："+loginLog,e);
		}
	}

	public void setJmsTemplate(JmsTemplate jmsTemplate) {
		this.jmsTemplate = jmsTemplate;
	}
	public void setNotifyQueue(Destination notifyQueue) {
		this.userLoginQueue = notifyQueue;
	}
}

LogSenderServiceImpl.java

package com.fei.service.log;

import java.util.Date;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.stereotype.Service;

import com.fei.dto.LoginLogDto;
import com.fei.dto.NewsVisitLogDto;
import com.fei.service.jmsproducer.NewsVisitLogProducer;
import com.fei.service.jmsproducer.UserLoginLogProducer;

@Service
public class LogSenderServiceImpl implements LogSenderService{

	@Autowired
	private UserLoginLogProducer userLoginLogProducer;
	
	@Autowired
	private NewsVisitLogProducer newsVisitLogProducer;
	
	public void send(LoginLogDto loginLogDto) {
		userLoginLogProducer.sendMessage(loginLogDto);
	}

	public void send(NewsVisitLogDto newsVisitLogDto) {
		newsVisitLogProducer.sendMessage(newsVisitLogDto);
	}

}

生產者專案中的測試例子

TestUserLoginLogProducer.java

package com.fei.service.jmsproducer;

import java.util.Date;

import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.springframework.context.support.ClassPathXmlApplicationContext;

import com.fei.dto.LoginLogDto;
import com.fei.dto.NewsVisitLogDto;
import com.fei.service.log.LogSenderService;

public class TestUserLoginLogProducer {

	private ClassPathXmlApplicationContext context ;
	
	@Before
	public void before(){
		context = new ClassPathXmlApplicationContext("spring/applicationContext.xml");
		context.start();
	}
	@After
	public void after(){
		context.stop();
	}
	@Test
	public void testSendMessage(){
		LoginLogDto u1 = new LoginLogDto();
		u1.setUserName("tom06");
		u1.setIp("192.168.6.95");
		u1.setLoginTime(new Date());
		
		LoginLogDto u2 = new LoginLogDto();
		u2.setUserName("jack06");
		u2.setIp("192.168.6.98");
		u2.setLoginTime(new Date());
		
		LogSenderService service = (LogSenderService)context.getBean("logSenderServiceImpl");
		service.send(u1);
		service.send(u2);
	}
	@Test
	public void testSendMessage2(){
		NewsVisitLogDto d1 = new NewsVisitLogDto();
		d1.setIp("192.168.22.33");
		d1.setNewsId("112244");
		d1.setVisitDate(new Date());
		
		LogSenderService service = (LogSenderService)context.getBean("logSenderServiceImpl");
		service.send(d1);
	}
}

執行testSendMessage()，這樣就把資料傳送到ActiveMq了。

再來看看消費者的監聽類

UserLoginLogListener.java

package com.fei.jmsconsumer;

import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.ObjectMessage;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import com.fei.dto.LoginLogDto;

@Service
public class UserLoginLogListener  implements MessageListener{

	private static Logger logger = LoggerFactory.getLogger(UserLoginLogListener.class);
	
	public void onMessage(Message message) {
		try {
			ObjectMessage objectMessage = ((ObjectMessage) message);
			LoginLogDto loginLogDto = (LoginLogDto)objectMessage.getObject();
			//呼叫其它service進行資料處理，這裡直接列印下
			logger.info("使用者登入:"+loginLogDto.getUserName() +"   "+loginLogDto.getIp()+"   "+loginLogDto.getLoginTime());
			System.out.println("使用者登入:"+loginLogDto.getUserName() +"   "+loginLogDto.getIp()+"   "+loginLogDto.getLoginTime());
		} catch (Exception e) {
			e.printStackTrace();
		}
		
		
	}

}

把jms_spring_consumer部署到tomcat，啟動，就可以看到列印日誌了

注意事項，必須要有日誌配置，比如logback.xml，同時web.xml中必須配有listener

<?xml version="1.0" encoding="UTF-8"?>
<web-app xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns="http://java.sun.com/xml/ns/javaee" xsi:schemaLocation="http://java.sun.com/xml/ns/javaee http://java.sun.com/xml/ns/javaee/web-app_2_5.xsd" id="WebApp_ID" version="2.5">
 <context-param>
    <param-name>contextConfigLocation</param-name>
    <param-value>classpath:spring/applicationContext*.xml</param-value>
  </context-param>
  
  <listener>
    <listener-class>org.springframework.web.context.ContextLoaderListener</listener-class>
  </listener>
  <listener>
    <listener-class>
            org.springframework.web.util.IntrospectorCleanupListener
        </listener-class>
  </listener>
</web-app>

這裡再貼下jms_spring_produce中的jms_applicationContext.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:jee="http://www.springframework.org/schema/jee"
	xmlns:tx="http://www.springframework.org/schema/tx" xmlns:context="http://www.springframework.org/schema/context"
	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-2.5.xsd http://www.springframework.org/schema/jee http://www.springframework.org/schema/jee/spring-jee-2.5.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-2.5.xsd">

	<bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
		<!-- 此處地址需要與啟動的jms服務中的配置檔案中activemq.xml中的 <transportConnector name="openwire" 
			uri="tcp://127.0.0.1:61616"中的埠指定一致/>配置 -->
		<property name="brokerURL" value="tcp://127.0.0.1:61616" />

 <!-- <property name="userName" value="system" /> <property name="password" value="manager"/> -->
		<!-- 對PERSISTENT的訊息進行同步傳送(NON_PERSISTENT訊息預設非同步傳送) -->
	       <property name="useAsyncSend" value="true" /> 
	</bean>

	<bean id="cachingConnectionFactory"
		class="org.springframework.jms.connection.CachingConnectionFactory">
		<property name="targetConnectionFactory" ref="connectionFactory" />
		<property name="sessionCacheSize" value="10" />
	</bean>
	<!-- Spring JMS Template -->
	<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
		<property name="connectionFactory" ref="cachingConnectionFactory" />
		<property name="deliveryPersistent" value="false" /> 
		<property name="priority" value="9" />
	</bean>
    <!-- 使用者登入 -->
	<bean id="userLoginQueue" class="org.apache.activemq.command.ActiveMQQueue">
		<constructor-arg value="queue.user.login" />
	</bean>
    <!-- 新聞點選 -->
     <bean id="newsVisitQueue" class="org.apache.activemq.command.ActiveMQQueue">
		<constructor-arg value="queue.news.visit" />
	</bean>
</beans>

消費者jms_spring_consumer中的jms_applicationContext.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:jee="http://www.springframework.org/schema/jee"
	xmlns:tx="http://www.springframework.org/schema/tx" xmlns:context="http://www.springframework.org/schema/context"
	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-2.5.xsd http://www.springframework.org/schema/jee http://www.springframework.org/schema/jee/spring-jee-2.5.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-2.5.xsd">

	<bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
		<!-- 此處地址需要與啟動的jms服務中的配置檔案中activemq.xml中的 <transportConnector name="openwire" 
			uri="tcp://127.0.0.1:61616"中的埠指定一致/>配置 -->
		<property name="brokerURL" value="tcp://127.0.0.1:61616" />

 <!-- <property name="userName" value="system" /> <property name="password" value="manager"/> -->
		<!-- 對PERSISTENT的訊息進行同步傳送(NON_PERSISTENT訊息預設非同步傳送) -->
	       <property name="useAsyncSend" value="true" /> 
	</bean>

	<bean id="cachingConnectionFactory"
		class="org.springframework.jms.connection.CachingConnectionFactory">
		<property name="targetConnectionFactory" ref="connectionFactory" />
		<property name="sessionCacheSize" value="10" />
	</bean>
	<!-- Spring JMS Template -->
	<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
		<property name="connectionFactory" ref="cachingConnectionFactory" />
		<property name="deliveryPersistent" value="false" /> 
		<property name="priority" value="9" />
	</bean>
    <!-- 使用者登入 -->
	<bean id="userLoginQueue" class="org.apache.activemq.command.ActiveMQQueue">
		<constructor-arg value="queue.user.login" />
	</bean>
	<bean id="userLoginQueueContainer"
		class="org.springframework.jms.listener.DefaultMessageListenerContainer">
		<property name="connectionFactory" ref="cachingConnectionFactory" />
		<property name="destination" ref="userLoginQueue" />
		<property name="messageListener" ref="userLoginLogListener" />
		<property name="concurrentConsumers" value="5" />
		<property name="maxConcurrentConsumers" value="10" />
	</bean>
    <!-- 新聞點選 -->
     <bean id="newsVisitQueue" class="org.apache.activemq.command.ActiveMQQueue">
		<constructor-arg value="queue.news.visit" />
	</bean>
    <bean id="newsVisitQueueContainer"
		class="org.springframework.jms.listener.DefaultMessageListenerContainer">
		<property name="connectionFactory" ref="cachingConnectionFactory" />
		<property name="destination" ref="newsVisitQueue" />
		<property name="messageListener" ref="newsVisitLogListener" />
		<property name="concurrentConsumers" value="5" />
		<property name="maxConcurrentConsumers" value="10" />
	</bean>
	
</beans>

spring的配置，有很多引數，也有很多負責的功能，這裡就不一一介紹了。

上面總共給出了4個專案jms_produce、jms_consumer、jms_spring_produce、jms_spring_consumer,這裡打包上傳了，大家可以下載看看

JMS+activeMq+spring簡單學習及模擬處理新聞日誌場景

下載安裝執行activeMq

JMS基本概念

Queue的簡單例子1

Queue的簡單例子2

Topic的簡單例子

JMS的一些深入知識

JMS+ActiveMq+Spring整合模擬處理新聞日誌場景

JMS+activeMq+spring簡單學習及模擬處理新聞日誌場景

Spring Boot學習筆記-錯誤處理及自定義

Spring + JMS + ActiveMQ實現簡單的訊息佇列（監聽器非同步實現）

Spring Boot學習——統一異常處理

Spring+Hibernate異常及錯誤處理——彙總

spring-boot使用AOP統一處理日誌

Spring-Batch學習總結（1）——重要概念，環境搭建，名詞解釋，第一個項目及異常處理

機器學習資料處理時label錯位對未來資料做預測機器學習經典模型簡單使用及歸一化（標準化）影響

Spring+SpringMVC+MyBatis深入學習及搭建(三)——MyBatis全局配置文件解析

Spring+SpringMVC+MyBatis深入學習及搭建(四)——MyBatis輸入映射與輸出映射

Spring+SpringMVC+MyBatis深入學習及搭建(八)——MyBatis查詢緩存

Spring+SpringMVC+MyBatis深入學習及搭建(十)——MyBatis逆向工程

Spring+SpringMVC+MyBatis深入學習及搭建(十一)——SpringMVC架構

Spring+SpringMVC+MyBatis深入學習及搭建(十四)——SpringMVC和MyBatis整合

深入淺出JMS(四)--Spring和ActiveMQ整合的完整實例

Spring+SpringMVC+MyBatis深入學習及搭建(二)——MyBatis原始Dao開發和mapper代理開發

Golang學習-第一篇 Golang的簡單介紹及Windows環境下安裝、部署

Spring+SpringMVC+MyBatis深入學習及搭建(四)——MyBatis輸入映射與輸出映射（轉發同上）

Spring Boot學習——AOP編程的簡單實現

spring+activemq中多個consumer同時處理消息時遇到的性能問題

JMS+activeMq+spring簡單學習及模擬處理新聞日誌場景

下載安裝執行activeMq

JMS基本概念

Queue的簡單例子1

Queue的簡單例子2

Topic的簡單例子

JMS的一些深入知識

JMS+ActiveMq+Spring整合模擬處理新聞日誌場景

相關推薦