JMS

較早以前，也就是沒有JMS的那個時候，很多應用系統存在一些缺陷：

1.通訊的同步性

client端發起呼叫後，必須等待server處理完成並返回結果後才能繼續執行

2.client 和 server 的生命週期耦合太高

client程序和server服務程序都必須可用，如果server出現問題或者網路故障，那麼client端會收到異常

3.點對點通訊

client端的一次呼叫只能傳送給某一個單獨的服務物件，無法一對多

JMS，即Java Message Service，通過面向訊息中介軟體（MOM：Message Oriented Middleware）的方式很好的解決了上面的問題。大致的過程是這樣的：傳送者把訊息傳送給訊息伺服器，訊息伺服器將訊息存放在若干佇列/主題中，在合適的時候，訊息伺服器會將訊息轉發給接受者。在這個過程中，傳送和接受是非同步的，也就是傳送無需等待，而且傳送者和接受者的生命週期也沒有必然關係；在pub/sub模式下，也可以完成一對多的通訊，即讓一個訊息有多個接受者。

JMS

需要注意的是，JMS只是定義了Java訪問訊息中介軟體的介面，其實就是在包javax.jms中，你會發現這個包下除了異常定義，其他都是interface。

JMS只給出介面，然後由具體的中介軟體去實現，比如ActiveMQ就是實現了JMS的一種Provider，還有阿里巴巴的RocketMQ。這些訊息中介軟體都符合JMS規範。說起規範，自然要定義一些術語：

Provider/MessageProvider：生產者

Consumer/MessageConsumer：消費者

PTP：Point To Point，點對點通訊訊息模型

Pub/Sub：Publish/Subscribe，釋出訂閱訊息模型

Queue：佇列，目標型別之一，和PTP結合

Topic：主題，目標型別之一，和Pub/Sub結合

ConnectionFactory：連線工廠，JMS用它建立連線

Connnection：JMS Client到JMS Provider的連線

Destination：訊息目的地，由Session建立

Session：會話，由Connection建立，實質上就是傳送、接受訊息的一個執行緒，因此生產者、消費者都是Session建立的

初步來看，Session非常核心，因為很多東西都是它建立的。

ActiveMQ是Apache出品的開源專案，非常流行的訊息中介軟體，是一種在分散式系統中應用程式藉以傳遞訊息的媒介，完全支援JMS1.1和J2EE1.4規範的JMS Provider實現。可以說要掌握訊息中介軟體，需要從ActiveMQ開始，要掌握更加強大的RocketMQ，也需要ActiveMQ的基礎。

特點：

1、支援多種語言編寫客戶端,語言: Java, C, C++, C#, Ruby, Perl, Python, PHP。
2、對spring的支援，很容易內嵌到使用Spring的系統裡面去,而且也支援Spring2.0的特性
3、支援多種傳輸協議：TCP,SSL,NIO,UDP等
4、支援AJAX
5. 通過了常見J2EE伺服器(如 Geronimo,JBoss 4, GlassFish,WebLogic)的測試,其中通過JCA 1.5 resource adaptors的配置,可以讓ActiveMQ可以自動的部署到任何相容J2EE 1.4 商業伺服器上
6. 支援通過JDBC和journal提供高速的訊息持久化
7. 從設計上保證了高效能的叢集,客戶端-伺服器,點對點
8. 支援與Axis的整合
9. 可以很容易得呼叫內嵌JMS provider,進行測試

訊息形式：
1、點對點（queue）一個生產者對應一個消費者
2、一對多（釋出訂閱模式）（topic）一個生產者產生訊息並進行傳送後，可以由多個消費者進行接收。

JMS定義了五種不同的訊息正文格式，以及呼叫的訊息型別，允許你傳送並接收以一些不同形式的資料，提供現有訊息格式的一些級別的相容性。

· StreamMessage -- Java原始值的資料流

· MapMessage--一套名稱-值對

· TextMessage--一個字串物件

· ObjectMessage--一個序列化的 Java物件

· BytesMessage--一個位元組的資料流

安裝：

在linux下安裝，需要先安裝好JDK

使用方法：

JMS訊息傳送模式

在點對點或佇列模型下，一個生產者向一個特定的佇列釋出訊息，一個消費者從該佇列中讀取訊息。這裡，生產者知道消費者的佇列，並直接將訊息傳送到消費者的佇列。這種模式被概括為：只有一個消費者將獲得訊息。生產者不需要在接收者消費該訊息期間處於執行狀態，接收者也同樣不需要在訊息傳送時處於執行狀態。每一個成功處理的訊息都由接收者簽收。

釋出者／訂閱者模型支援向一個特定的訊息主題釋出訊息。0或多個訂閱者可能對接收來自特定訊息主題的訊息感興趣。在這種模型下，釋出者和訂閱者彼此不知道對方。這種模式好比是匿名公告板。這種模式被概括為：多個消費者可以獲得訊息.在釋出者和訂閱者之間存在時間依賴性。釋出者需要建立一個訂閱（subscription），以便客戶能夠購訂閱。訂閱者必須保持持續的活動狀態以接收訊息，除非訂閱者建立了持久的訂閱。在那種情況下，在訂閱者未連線時釋出的訊息將在訂閱者重新連線時重新發布。

JMS應用程式介面

ConnectionFactory 介面（連線工廠）

使用者用來建立到JMS提供者的連線的被管物件。JMS客戶通過可移植的介面訪問連線，這樣當下層的實現改變時，程式碼不需要進行修改。管理員在JNDI名字空間中配置連線工廠，這樣，JMS客戶才能夠查詢到它們。根據訊息型別的不同，使用者將使用佇列連線工廠，或者主題連線工廠。

Connection 介面（連線）

連線代表了應用程式和訊息伺服器之間的通訊鏈路。在獲得了連線工廠後，就可以建立一個與JMS提供者的連線。根據不同的連線型別，連線允許使用者建立會話，以傳送和接收佇列和主題到目標。

Destination 介面（目標）

目標是一個包裝了訊息目標識別符號的被管物件，訊息目標是指訊息釋出和接收的地點，或者是佇列，或者是主題。JMS管理員建立這些物件，然後使用者通過JNDI發現它們。和連線工廠一樣，管理員可以建立兩種型別的目標，點對點模型的佇列，以及釋出者／訂閱者模型的主題。

MessageConsumer 介面（訊息消費者）

由會話建立的物件，用於接收發送到目標的訊息。消費者可以同步地（阻塞模式），或非同步（非阻塞）接收佇列和主題型別的訊息。

MessageProducer 介面（訊息生產者）

由會話建立的物件，用於傳送訊息到目標。使用者可以建立某個目標的傳送者，也可以建立一個通用的傳送者，在傳送訊息時指定目標。

Message 介面（訊息）

是在消費者和生產者之間傳送的物件，也就是說從一個應用程式創送到另一個應用程式。一個訊息有三個主要部分：

訊息頭（必須）：包含用於識別和為訊息尋找路由的操作設定。

一組訊息屬性（可選）：包含額外的屬性，支援其他提供者和使用者的相容。可以建立定製的欄位和過濾器（訊息選擇器）。

一個訊息體（可選）：允許使用者建立五種型別的訊息（文字訊息，對映訊息，位元組訊息，流訊息和物件訊息）。

訊息介面非常靈活，並提供了許多方式來定製訊息的內容。

Session 介面（會話）

表示一個單執行緒的上下文，用於傳送和接收訊息。由於會話是單執行緒的，所以訊息是連續的，就是說訊息是按照發送的順序一個一個接收的。會話的好處是它支援事務。如果使用者選擇了事務支援，會話上下文將儲存一組訊息，直到事務被提交才傳送這些訊息。在提交事務之前，使用者可以使用回滾操作取消這些訊息。一個會話允許使用者建立訊息生產者來發送訊息，建立訊息消費者來接收訊息。

1.訊息佇列

把ActiveMQ依賴的jar包新增到工程中。

activemq-all-5.12.0.jar

使用maven工程，則新增jar包的依賴：

<groupId>org.apache.activemq</groupId>

<artifactId>activemq-all</artifactId>

</dependency>

Producer

public class QueueSender {

public static void main(String[] args) {

//建立一個連線工廠

ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.168:61616");

try {

//從工廠物件中獲得連線

Connection connection = connectionFactory.createConnection();

//開啟連線

connection.start();

connection.createSession(paramA, paramB)

A）paramA設定為true時：

paramB的值忽略， acknowledgment mode被jms伺服器設定 SESSION_TRANSACTED 。

當一個事務被提交的時候，訊息確認就會自動發生。

B） paramA設定為false時：

Session.AUTO_ACKNOWLEDGE為自動確認，當客戶成功的從receive方法返回的時候，或者從

MessageListener.onMessage方法成功返回的時候，會話自動確認客戶收到的訊息。

Session.CLIENT_ACKNOWLEDGE 為客戶端確認。客戶端接收到訊息後，必須呼叫javax.jms.Message的

acknowledge方法。jms伺服器才會刪除訊息。（預設是批量確認）

//開啟一個回話，第一個引數指定不使用事務，第二個引數指定客戶端接收訊息的確認方式

Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

//建立一目的地Queue或者是Topic

Queue queue = session.createQueue("mytestqueue");

//建立一個生產者

MessageProducer producer = session.createProducer(queue);

//建立message

TextMessage message = new ActiveMQTextMessage();

message.setText("hello");

//傳送訊息

producer.send(message);

//關閉

producer.close();

session.close();

connection.close();

} catch (JMSException e) {

// TODO Auto-generated catch block

e.printStackTrace();

}

Consumer

消費者有兩種消費方法：：

1、同步消費。通過呼叫消費者的receive方法從目的地中顯式提取訊息。receive方法可以一直阻塞到訊息到達。

2、非同步消費。客戶可以為消費者註冊一個訊息監聽器，以定義在訊息到達時所採取的動作。

實現MessageListener介面，在MessageListener（）方法中實現訊息的處理邏輯。

同步消費

public class QueueConsumer {

public static void main(String[] args) {

//建立一連線工廠

ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.168:61616");

try {

//建立一個連線

Connection connection = connectionFactory.createConnection();

//開啟連線

connection.start();

//建立一個回話

Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

//建立一個目的地Destination

Queue queue = session.createQueue("mytestqueue");

//建立一個消費者

MessageConsumer consumer = session.createConsumer(queue);

while(true) {

//設定接收者接收訊息的時間，為了便於測試，這裡定為100s

Message message = consumer.receive(100000);

if (message != null) {

System.out.println(message);

} else {

//超時結束

break;

}

consumer.close();

session.close();

connection.close();

} catch (Exception e) {

e.printStackTrace();

}

1.非同步消費

public class QueueConsumer {

public static void main(String[] args) {

//建立一連線工廠

ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.168:61616");

try {

//建立一個連線

Connection connection = connectionFactory.createConnection();

//開啟連線

connection.start();

//建立一個回話

Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

//建立一個目的地Destination

Queue queue = session.createQueue("mytestqueue");

//建立一個消費者

MessageConsumer consumer = session.createConsumer(queue);

consumer.setMessageListener(new MessageListener() {

@Override

public void onMessage(Message message) {

if (message instanceof TextMessage) {

String text = "";

try {

text = ((TextMessage)message).getText();

} catch (JMSException e) {

// TODO Auto-generated catch block

e.printStackTrace();

}

System.out.println(text);

}

});

System.in.read();

//關閉

consumer.close();

session.close();

connection.close();

} catch (Exception e) {

e.printStackTrace();

}

釋出者／訂閱者
1. Producer

public class TopicProducer {

public static void main(String[] args) {

//建立連線工廠

ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.168:61616");

try {

//建立連線

Connection connection = connectionFactory.createConnection();

//開啟連線

connection.start();

//建立一個回話

Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

//建立一個Destination，queue或者Topic

Topic topic = session.createTopic("mytopic");

//建立一個生成者

MessageProducer producer = session.createProducer(topic);

//建立一個訊息

TextMessage textMessage = new ActiveMQTextMessage();

textMessage.setText("hello my topic");

//傳送訊息

producer.send(textMessage);

//關閉

producer.close();

session.close();

connection.close();

} catch (JMSException e) {

// TODO Auto-generated catch block

e.printStackTrace();

}

Consumer

public class TopicConsumer {

public static void main(String[] args) {

//建立連線工廠

ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.168:61616");

try {

//建立連線

Connection connection = connectionFactory.createConnection();

connection.start();

//建立一個會話

Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

//建立一個目標

Destination destination = session.createTopic("mytopic");

//建立一個消費者

MessageConsumer consumer = session.createConsumer(destination);

//接收訊息

consumer.setMessageListener(new MessageListener() {

@Override

public void onMessage(Message message) {

System.out.println(message);

}

});

//暫停

System.in.read();

//關閉

consumer.close();

session.close();

connection.close();

} catch (Exception e) {

// TODO Auto-generated catch block

e.printStackTrace();

}

ActiveMQ整合Spring
1. 配置ConnectionFactory

<?xml version="1.0" encoding="UTF-8"?>

<beans xmlns="http://www.springframework.org/schema/beans"

xmlns:context="http://www.springframework.org/schema/context" xmlns:p="http://www.springframework.org/schema/p"

xmlns:aop="http://www.springframework.org/schema/aop" xmlns:tx="http://www.springframework.org/schema/tx"

xmlns:jms="http://www.springframework.org/schema/jms" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

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/aop http://www.springframework.org/schema/aop/spring-aop-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/jms http://www.springframework.org/schema/jms/spring-jms-4.0.xsd

http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util-4.0.xsd">

</bean>

<bean id="connectionFactory"

class="org.springframework.jms.connection.SingleConnectionFactory">

</bean>

</beans>

1. 配置生產者
  1. Spring配置檔案

</bean>

<constructor-arg>

<value>queue</value>

</constructor-arg>

</bean>

<constructor-arg value="topic" />

</bean>

1. 傳送訊息

public class Producer {

public static void main(String[] args) {

//建立spring容器

ApplicationContext applicationContext = new ClassPathXmlApplicationContext("applicationContext.xml");

//獲得JmsTemplate物件

JmsTemplate template = (JmsTemplate) applicationContext.getBean("jmsTemplate");

//獲得Destination

ActiveMQQueue queue = (ActiveMQQueue) applicationContext.getBean("queueDestination");

//傳送訊息

template.send(queue, new MessageCreator() {

@Override

public Message createMessage(Session session) throws JMSException {

return session.createTextMessage("hello");

}

});

}

1. 配置消費者

那麼消費者是通過Spring為我們封裝的訊息監聽容器MessageListenerContainer實現的，它負責接收資訊，並把接收到的資訊分發給真正的MessageListener進行處理。每個消費者對應每個目的地都需要有對應的MessageListenerContainer。

對於訊息監聽容器而言，除了要知道監聽哪個目的地之外，還需要知道到哪裡去監聽，也就是說它還需要知道去監聽哪個JMS伺服器，這是通過在配置MessageConnectionFactory的時候往裡面注入一個ConnectionFactory來實現的。

所以在配置一個MessageListenerContainer的時候有三個屬性必須指定：

一個是表示從哪裡監聽的ConnectionFactory
一個是表示監聽什麼的Destination；
一個是接收到訊息以後進行訊息處理的MessageListener。

常用的MessageListenerContainer實現類是DefaultMessageListenerContainer。

1. MessageListener

public class MyMessageListener implements MessageListener {

@Override

public void onMessage(Message message) {

System.out.println(message);

}

1. Spring配置檔案

<?xml version="1.0" encoding

AcitveMQ的介紹、虛擬機器linux安裝activeMQ、詳細使用教程

JMS

特點：

安裝：

使用方法：

AcitveMQ的介紹、虛擬機器linux安裝activeMQ、詳細使用教程

在虛擬機器中安裝Ubuntu，詳細教程

linux虛擬機器下安裝activemq及出現的奇葩問題

最詳細的虛擬機器linux安裝教程

虛擬機器Linux安裝

vm虛擬機器linux安裝vm tools

虛擬機器linux安裝wget小問題

VMware Workstation pro14 虛擬機器下安裝CentOS6.5圖文教程

虛擬機器下安裝CentOS6.5系統教程

虛擬機器VirtualBox安裝windows作業系統(圖文教程)

JAVAWEB開發之——Linux命令實戰、虛擬機器和centos的安裝以及聯網、常用命令、vi編輯以及軟體的安裝、sftp的使用

JavaEE開發Linux系列之《一、虛擬機器中centos 6.5minimal最小化安裝與ip網路配置》

Linux centosVMware NFS介紹、NFS服務端安裝配置、NFS配置選項

linux虛擬機器的安裝與封裝

linux下怎麼編寫指令碼讓虛擬機器自動安裝系統和安裝軟體

linux虛擬機器的安裝和管理

Linux運維入門～14.虛擬機器的安裝及使用

redis在Linux虛擬機器上安裝了，在windows下無法訪問的問題

Linux——虛擬機器的安裝

在虛擬機器中安裝Linux系統CentOS7詳細教程

AcitveMQ的介紹、虛擬機器linux安裝activeMQ、詳細使用教程

JMS

特點：

安裝：

使用方法：

相關推薦