ACTIVEMQ設置Timestamp和jms簡介


在默認情況下,向activemq的broker中發送消息時,messageid是系統自己生成的,通常和自己所使用計算機名相關。Timestamp是計算機的系統時間。

更改Timestamp:

TextMessage messageForSend = null;    //發送的消息
String messageinfo = (String)messageconnect.receiveMessage();
if(messageinfo==null)
continue;
messageForSend = session.createTextMessage(messageinfo);

messSender.setDisableMessageTimestamp(true);           //關鍵要設置一下
messageForSend.setJMSTimestamp(System.currentTimeMillis());   //也可以改為其他的時間
messSender.send(messageForSend);

 

JMS
     JMS(Java Message Service) 即Java消息服務。它提供標准的產生、發送、接收消息的接口簡化企業應用的開發。它支持兩種消息通信模型:點到點(point-to-point)(P2P)模型和發布/訂閱(Pub/Sub)模型。P2P 模型規定了一個消息只能有一個接收者;Pub/Sub 模型允許一個消息可以有多個接收者。
    對於點到點模型,消息生產者產生一個消息后,把這個消息發送到一個Queue(隊列)中,然后消息接收者再從這個Queue中讀取,一旦這個消息被一個接收者讀取之后,它就在這個Queue中消失了,所以一個消息只能被一個接收者消費。
    與點到點模型不同,發布/訂閱模型中,消息生產者產生一個消息后,把這個消息發送到一個Topic中,這個Topic可以同時有多個接收者在監聽,當一個消息到達這個Topic之后,所有消息接收者都會收到這個消息。簡單的講,點到點模型和發布/訂閱模型的區別就是前者是一對一,后者是一對多。
幾個重要概念
Destination:消息發送的目的地,也就是前面說的Queue和Topic。創建好一個消息之后,只需要把這個消息發送到目的地,消息的發送者就可以繼續做自己的事情,而不用等待消息被處理完成。至於這個消息什么時候,會被哪個消費者消費,完全取決於消息的接受者。
    Message:從字面上就可以看出是被發送的消息。它有下面幾種類型:
        StreamMessage:Java 數據流消息,用標准流操作來順序的填充和讀取。
        MapMessage:一個Map類型的消息;名稱為 string 類型,而值為 Java 的基本類型。
        TextMessage:普通字符串消息,包含一個String。
        ObjectMessage:對象消息,包含一個可序列化的Java 對象
        BytesMessage:二進制數組消息,包含一個byte[]。
        XMLMessage:  一個XML類型的消息。
    最常用的是TextMessage和ObjectMessage。
   Session:與JMS提供者所建立的會話,通過Session我們才可以創建一個Message。
   Connection:與JMS提供者建立的一個連接。可以從這個連接創建一個會話,即Session。
   ConnectionFactory:那如何創建一個Connection呢?這就需要下面講到的ConnectionFactory了。通過這個工廠類就可以得到一個與JMS提供者的連接,即Conection。
   Producer:消息的生產者,要發送一個消息,必須通過這個生產者來發送。
   MessageConsumer:與生產者相對應,這是消息的消費者或接收者,通過它來接收一個消息。
   前面多次提到JMS提供者,因為JMS給我們提供的只是一系列接口,當我們使用一個JMS的時候,還是需要一個第三方的提供者,它的作用就是真正管理這些Connection,Session,Topic和Queue等。通過下面這個簡圖可以看出上面這些概念的關系。
  ConnectionFactory---->Connection--->Session--->Message
Destination + Session------------------------------------>Producer
Destination + Session------------------------------------>MessageConsumer
 
    那么可能有人會問: ConnectionFactory和Destination 從哪兒得到?這就和JMS提供者有關了. 如果在一個JavaEE環境中, 可以通過JNDI查找得到, 如果在一個非JavaEE環境中, 那只能通過JMS提供者提供給我們的接口得到了.這里我們選擇ActiveMQ,結合一個例子讓大家深入理解上面的那個簡圖:

首先需要得到ConnectionFactoy和Destination,這里創建一個一對一的Queue作為Destination。
  ConnectionFactory factory = new ActiveMQConnectionFactory("vm://localhost");
  Queue queue = new ActiveMQQueue("testQueue");

然后又ConnectionFactory創建一個Connection, 再啟動這個Connection:
  Connection connection = factory.createConnection();
  connection.start();

接下來需要由Connection創建一個Session:
  Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE)
    現在暫且不用管參數的含義, 以后會詳細講到.

下面就可以創建Message了,這里創建一個TextMessage。
  Message message = session.createTextMessage("Hello JMS!");

要想把剛才創建的消息發送出去,需要由Session和Destination創建一個消息生產者:
  MessageProducer producer = session.createProducer(queue);

下面就可以發送剛才創建的消息了:
  producer.send(message);

消息發送完成之后,我們需要創建一個消息消費者來接收這個消息:
  MessageConsumer comsumer = session.createConsumer(queue);
  Message recvMessage = comsumer.receive();

消息消費者接收到這個消息之后,就可以得到它的內容:
  System.out.println(((TextMessage)recvMessage).getText());
至此,一個簡單的JMS例子就完成了。

消息的消費者接收消息可以采用兩種方式
  1、consumer.receive() 或 consumer.receive(int timeout);
  2、注冊一個MessageListener。
  采用第一種方式,消息的接收者會一直等待下去,直到有消息到達,或者超時。后一種方式會注冊一個監聽器,當有消息到達的時候,會回調它的onMessage()方法。下面舉例說明:
 MessageConsumer comsumer = session.createConsumer(queue);
 comsumer.setMessageListener(new MessageListener(){
            @Override
            public void onMessage(Message m) {
                TextMessage textMsg = (TextMessage) m;
                try {
                    System.out.println(textMsg.getText());
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
           
        });
        

Queue
Queue實現的是點到點模型,如果啟動多個消費者共同監聽一個Queue,然后循環給這個Queue中發送多個消息,結果每個消息只會被消費一次,我們無法確定一個消息最終會被哪一個消費者消費。
Topic:
與Queue不同的是,Topic實現的是發布/訂閱模型,啟動多個消費者共同監聽一個Topic,然后循環給這個Topic中發送多個消息。結果每一個消息都會被所有的消費者消費。

消息頭

一個消息對象分為三部分:消息頭(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
一個字符串用來唯一標示一個消息,J主要是用來關聯多個Message,例如需要回復一個消息的時候,通常把回復的消息的JMSCorrelationID設置為原來消息的ID。

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

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

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

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

除了Header,消息發送者可以添加一些屬性(Properties)。這些屬性可以是應用自定義的屬性,JMS定義的屬性和JMS提供者定義的屬性。我們通常只適用自定義的屬性。
JMSReplyTo
在下面的例子中,首先創建兩個Queue,發送者給一個Queue發送,接收者接收到消息之后給另一個Queue回復一個Message,然后再創建一個消費者來接受所回復的消息。
public class MessageSendReceiveAndReply {

public static void main(String[] args) throws Exception {
ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("vm://localhost");
Connection connection = factory.createConnection();
connection.start();

//消息發送到這個Queue
Queue queue = new ActiveMQQueue("testQueue");

//消息回復到這個Queue
Queue replyQueue = new ActiveMQQueue("replyQueue");

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

//創建一個消息,並設置它的JMSReplyTo為replyQueue。
Message message = session.createTextMessage("Andy");
message.setJMSReplyTo(replyQueue);

MessageProducer producer = session.createProducer(queue);
producer.send(message);

//消息的接收者
MessageConsumer comsumer = session.createConsumer(queue);
comsumer.setMessageListener(new MessageListener(){
public void onMessage(Message m) {
try {
//創建一個新的MessageProducer來發送一個回復消息。
MessageProducer producer = session.createProducer(m.getJMSReplyTo());
producer.send(session.createTextMessage("Hello " + ((TextMessage) m).getText()));
} catch (JMSException e1) {
e1.printStackTrace();
}
}

});

//這個接收者用來接收回復的消息
MessageConsumer comsumer2 = session.createConsumer(replyQueue);
comsumer2.setMessageListener(new MessageListener(){
public void onMessage(Message m) {
try {
System.out.println(((TextMessage) m).getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
});
}

}
首先消息生產者發送一個消息,內容為“Andy”, 然后消費者收到這個消息之后根據消息的JMSReplyTo,回復一個消息,內容為“Hello Andy‘。 最后在回復的Queue上創建一個接收回復消息的消費者,它輸出所回復的內容。
運行上面的程序,可以得到下面的輸出結果:Hello Andy
Selector
前面的例子中創建一個消息消費者使用的是:
sesssion.createConsumer(destination)
另外,還提供了另一種方式:
sesssion.createConsumer(destination, selector)
這里selector是一個字符串,用來過濾消息。也就是說,這種方式可以創建一個可以只接收特定消息的一個消費者。Selector的格式是類似於SQL-92的一種語法。可以用來比較消息頭信息和屬性。
下面的例子中,創建兩個消費者,共同監聽同一個Queue,但是它們的Selector不同,然后創建一個消息生產者,來發送多個消息。

public class JMSSelectorTest {

public static void main(String[] args) throws Exception {
ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("vm://localhost");
Connection connection = factory.createConnection();
connection.start();

Queue queue = new ActiveMQQueue("testQueue");

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

MessageConsumer comsumerA = session.createConsumer(queue, "receiver = 'A'");
comsumerA.setMessageListener(new MessageListener(){
public void onMessage(Message m) {
try {
System.out.println("ConsumerA get " + ((TextMessage) m).getText());
} catch (JMSException e1) { }
}
});

MessageConsumer comsumerB = session.createConsumer(queue, "receiver = 'B'");
comsumerB.setMessageListener(new MessageListener(){
public void onMessage(Message m) {
try {
System.out.println("ConsumerB get " + ((TextMessage) m).getText());
} catch (JMSException e) { }
}
});

MessageProducer producer = session.createProducer(queue);
for(int i=0; i<3; i++) {
String receiver = (i%3 == 0 ? "A" : "B");
TextMessage message = session.createTextMessage("Message" + i + ", receiver:" + receiver);
message.setStringProperty("receiver", receiver);
producer.send(message );
}
}
}

結果如下:
ConsumerA get Message0, receiver:A
ConsumerB get Message1, receiver:B
ConsumerB get Message2, receiver:B
可以看出,消息消費者只會取走它自己感興趣的消息。
TemporaryQueue和TemporaryTopic
TemporaryQueue和TemporaryTopic,從字面上就可以看出它們是“臨時”的目的地。可以通過Session來創建,例如:
TemporaryQueue replyQueue = session.createTemporaryQueue();  雖然它們是由Session來創建的,但是它們的生命周期卻是整個Connection。如果在一個Connection上創建了兩個Session,則一個Session創建的TemporaryQueue或TemporaryTopic也可以被另一個Session訪問。那如果這兩個Session是由不同的Connection創建,則一個Session創建的TemporaryQueue不可以被另一個Session訪問。另外,它們的主要作用就是用來指定回復目的地, 即作為JMSReplyTo。
在下面的例子中,先創建一個Connection,然后創建兩個Session,其中一個Session創建了一個TemporaryQueue,另一個Session在這個TemporaryQueue上讀取消息。

public class TemporaryQueueTest {

public static void main(String[] args) throws Exception {
ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("vm://localhost");
Connection connection = factory.createConnection();
connection.start();

Queue queue = new ActiveMQQueue("testQueue2");
final Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//使用session創建一個TemporaryQueue。
TemporaryQueue replyQueue = session.createTemporaryQueue();

//接收消息,並回復到指定的Queue中(即replyQueue)
MessageConsumer comsumer = session.createConsumer(queue);
comsumer.setMessageListener(new MessageListener(){
public void onMessage(Message m) {
try {
System.out.println("Get Message: " + ((TextMessage)m).getText());
MessageProducer producer = session.createProducer(m.getJMSReplyTo());
producer.send(session.createTextMessage("ReplyMessage"));
} catch (JMSException e) { }
}
});

//使用同一個Connection創建另一個Session,來讀取replyQueue上的消息。
Session session2 = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
MessageConsumer replyComsumer = session2.createConsumer(replyQueue);
replyComsumer.setMessageListener(new MessageListener(){
public void onMessage(Message m) {
try {
System.out.println("Get reply: " + ((TextMessage)m).getText());
} catch (JMSException e) { }
}
});

MessageProducer producer = session.createProducer(queue);
TextMessage message = session.createTextMessage("SimpleMessage");
message.setJMSReplyTo(replyQueue);
producer.send(message);
}
}

運行結果為:
Get Message: SimpleMessage
Get reply: ReplyMessage

如果將:
Session session2 = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
更改為:
Connection connection2 = factory.createConnection();
Session session2 = connection2.createSession(true, Session.AUTO_ACKNOWLEDGE);
就會得到類似於下面的異常:
Exception in thread "main" javax.jms.InvalidDestinationException: Cannot use a Temporary destination from another Connection。


免責聲明!

本站轉載的文章為個人學習借鑒使用,本站對版權不負任何法律責任。如果侵犯了您的隱私權益,請聯系本站郵箱yoyou2525@163.com刪除。



 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM