ActiveMQ消息簽收機制代碼實例詳解

更新時間：2020年02月21日 14:41:49 作者：流氓大隊長

這篇文章主要介紹了ActiveMQ消息簽收機制代碼實例解析,文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下

消費者客戶端成功接收一條消息的標志是：這條消息被簽收。

消費者客戶端成功接收一條消息一般包括三個階段：

1、消費者接收消息，也即從MessageConsumer的receive方法返回

2、消費者處理消息

3、消息被簽收

其中，第三階段的簽收可以有ActiveMQ發(fā)起，也可以由消費者客戶端發(fā)起，取決于Session是否開啟事務以及簽收模式的設置。

在帶事務的Session中，消費者客戶端事務提交之時，消息自動完成簽收。

在不帶事務的Session中，消息何時以及如何被簽收取決于Session的簽收模式設置

非事務Session可以設置如下幾種簽收模式：

1.Session.AUTO_ACKNOWLEDGE

當消息從MessageConsumer的receive方法返回或者從MessageListener接口的onMessage方法返回時，會話自動確認消息簽收

2.Session.CLIENT_ACKNOWLEDGE

需要消費者客戶端主動調用acknowledge方法簽收消息，這種模式實在Session層面進行簽收的，簽收一個已經消費的消息會自動的簽收這個Session已消費的所有消息：

例如一個消費者在一個Session中消費了5條消息，然后確認第3條消息，所有這5條消息都會被簽收

3.Session.DUPS_OK_ACKNOWLEDGE

這種方式允許JMS不必急于確認收到的消息,允許在收到多個消息之后一次完成確認,與Auto_AcKnowledge相比,這種確認方式在某些情況下可能更有效,因為沒有確認,當系統(tǒng)崩潰或者網絡出現故障的時候,消息可以被重新傳遞.

這種方式會引起消息的重復，但是降低了Session的開銷，所以只有客戶端能容忍重復的消息才可使用。(如果ActiveMQ再次傳送同一消息，那么消息頭中的JMSRedelivered將被設置為true)

帶事務session的案例

　　生產者

　　　　必須在生產完數據之后手動提交session

package com.wn.ddd;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;

public class Producter {
 public static void main(String[] args) throws JMSException {
  // ConnectionFactory ：連接工廠，JMS 用它創(chuàng)建連接
  ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,
    ActiveMQConnection.DEFAULT_PASSWORD, "tcp://127.0.0.1:61616");
  // JMS 客戶端到JMS Provider 的連接
  Connection connection = connectionFactory.createConnection();
  //啟動連接
  connection.start();
  // Session： 一個發(fā)送或接收消息的線程 false:代表不帶事務的session AUTO_ACKNOWLEDGE：代表自動簽收
  Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
  // Destination ：消息的目的地;消息發(fā)送給誰.
  // 獲取session注意參數值my-queue是Query的名字
  Queue queue = session.createQueue("my-queue");
  // MessageProducer：創(chuàng)建消息生產者
  MessageProducer producer = session.createProducer(queue);
  // 設置不持久化 PERSISTENT:代表持久化 NON_PERSISTENT：代表不持久化
  producer.setDeliveryMode(DeliveryMode.PERSISTENT);
  // 發(fā)送消息
  for (int i = 1; i <= 5; i++) {
   sendMsg(session, producer, i);
  }
  System.out.println("發(fā)送成功！");
  session.commit();
  session.close();
  connection.close();
 }
 /**
  * 在指定的會話上，通過指定的消息生產者發(fā)出一條消息
  *
  * @param session
  *   消息會話
  * @param producer
  *   消息生產者
  */
 public static void sendMsg(Session session, MessageProducer producer, int i) throws JMSException {
  // 創(chuàng)建一條文本消息
  TextMessage message = session.createTextMessage("Hello ActiveMQ！" + i);
  // 通過消息生產者發(fā)出消息
  producer.send(message);
 }
}

　　消費者

　　　　消費完數據之后必須手動提交session

package com.wn.ddd;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

public class JmsReceiver {
 public static void main(String[] args) throws JMSException {
  // ConnectionFactory ：連接工廠，JMS 用它創(chuàng)建連接
  ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,
    ActiveMQConnection.DEFAULT_PASSWORD, "tcp://127.0.0.1:61616");
  // JMS 客戶端到JMS Provider 的連接
  Connection connection = connectionFactory.createConnection();
  connection.start();
  // Session： 一個發(fā)送或接收消息的線程 true：表單開啟事務 AUTO_ACKNOWLEDGE：代表自動簽收
  Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
  // Destination ：消息的目的地;消息發(fā)送給誰.
  // 獲取session注意參數值xingbo.xu-queue是一個服務器的queue，須在在ActiveMq的console配置
  Queue queue = session.createQueue("my-queue");
  // 消費者，消息接收者
  MessageConsumer consumer = session.createConsumer(queue);
  while (true) {
   //receive():獲取消息
   TextMessage message = (TextMessage) consumer.receive();
   if (null != message) {
    System.out.println("收到消息：" + message.getText());
    session.commit();
   } else {
    break;
   }
  }
  //回收資源
  session.close();
  connection.close();
 }
}

不帶事務session的案例

　　1.自動簽收

　　2.手動簽收

　　　　生產者

package com.wn.ddd;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;

public class Producter {
 public static void main(String[] args) throws JMSException {
  // ConnectionFactory ：連接工廠，JMS 用它創(chuàng)建連接
  ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,
    ActiveMQConnection.DEFAULT_PASSWORD, "tcp://127.0.0.1:61616");
  // JMS 客戶端到JMS Provider 的連接
  Connection connection = connectionFactory.createConnection();
  //啟動連接
  connection.start();
  // Session： 一個發(fā)送或接收消息的線程 false:代表不帶事務的session AUTO_ACKNOWLEDGE：代表自動簽收
  /* Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);*/
  Session session = connection.createSession(Boolean.FALSE, Session.CLIENT_ACKNOWLEDGE);
  // Destination ：消息的目的地;消息發(fā)送給誰.
  // 獲取session注意參數值my-queue是Query的名字
  Queue queue = session.createQueue("my-queue");
  // MessageProducer：創(chuàng)建消息生產者
  MessageProducer producer = session.createProducer(queue);
  // 設置不持久化 PERSISTENT:代表持久化 NON_PERSISTENT：代表不持久化
  producer.setDeliveryMode(DeliveryMode.PERSISTENT);
  // 發(fā)送消息
  for (int i = 1; i <= 5; i++) {
   sendMsg(session, producer, i);
  }
  System.out.println("發(fā)送成功！");
  session.close();
  connection.close();
 }
 /**
  * 在指定的會話上，通過指定的消息生產者發(fā)出一條消息
  *
  * @param session
  *   消息會話
  * @param producer
  *   消息生產者
  */
 public static void sendMsg(Session session, MessageProducer producer, int i) throws JMSException {
  // 創(chuàng)建一條文本消息
  TextMessage message = session.createTextMessage("Hello ActiveMQ！" + i);
  // 通過消息生產者發(fā)出消息
  producer.send(message);
　　　　　message.acknowledge();　　//手動提交
　　} 
}

　　　　消費者

package com.wn.ddd;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import sun.plugin2.os.windows.SECURITY_ATTRIBUTES;

import javax.jms.*;

public class JmsReceiver {
 public static void main(String[] args) throws JMSException {
  // ConnectionFactory ：連接工廠，JMS 用它創(chuàng)建連接
  ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,
    ActiveMQConnection.DEFAULT_PASSWORD, "tcp://127.0.0.1:61616");
  // JMS 客戶端到JMS Provider 的連接
  Connection connection = connectionFactory.createConnection();
  connection.start();
  // Session： 一個發(fā)送或接收消息的線程 true：表單開啟事務 AUTO_ACKNOWLEDGE：代表自動簽收
  /*Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);*/
  Session session = connection.createSession(Boolean.FALSE, Session.CLIENT_ACKNOWLEDGE);
  // Destination ：消息的目的地;消息發(fā)送給誰.
  // 獲取session注意參數值xingbo.xu-queue是一個服務器的queue，須在在ActiveMq的console配置
  Queue queue = session.createQueue("my-queue");
  // 消費者，消息接收者
  MessageConsumer consumer = session.createConsumer(queue);
  while (true) {
   //receive():獲取消息
   TextMessage message = (TextMessage) consumer.receive();
   if (null != message) {
    System.out.println("收到消息：" + message.getText());
    message.acknowledge();　　//手動提交
   } else {
    break;
   }
  }
  //回收資源
  session.close();
  connection.close();
 }
}

以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支持腳本之家。

您可能感興趣的文章: