2. 程式人生 > 程式設計 >ActiveMQ訊息簽收機制程式碼例項詳解

ActiveMQ訊息簽收機制程式碼例項詳解

阿新 • • 發佈：2020-02-23

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

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

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

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

2、消費者處理訊息

3、訊息被簽收

其中，第三階段的簽收可以有ActiveMQ發起，也可以由消費者客戶端發起，取決於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相比,這種確認方式在某些情況下可能更有效,因為沒有確認,當系統崩潰或者網路出現故障的時候,訊息可以被重新傳遞.

這種方式會引起訊息的重複，但是降低了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 用它建立連線
  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： 一個傳送或接收訊息的執行緒 false:代表不帶事務的session AUTO_ACKNOWLEDGE：代表自動簽收
  Session session = connection.createSession(Boolean.TRUE,Session.AUTO_ACKNOWLEDGE);
  // Destination ：訊息的目的地;訊息傳送給誰.
  // 獲取session注意引數值my-queue是Query的名字
  Queue queue = session.createQueue("my-queue");
  // MessageProducer：建立訊息生產者
  MessageProducer producer = session.createProducer(queue);
  // 設定不持久化 PERSISTENT:代表持久化 NON_PERSISTENT：代表不持久化
  producer.setDeliveryMode(DeliveryMode.PERSISTENT);
  // 傳送訊息
  for (int i = 1; i <= 5; i++) {
   sendMsg(session,producer,i);
  }
  System.out.println("傳送成功！");
  session.commit();
  session.close();
  connection.close();
 }
 /**
  * 在指定的會話上，通過指定的訊息生產者發出一條訊息
  *
  * @param session
  *   訊息會話
  * @param producer
  *   訊息生產者
  */
 public static void sendMsg(Session session,MessageProducer producer,int i) throws JMSException {
  // 建立一條文字訊息
  TextMessage message = session.createTextMessage("Hello ActiveMQ！" + i);
  // 通過訊息生產者發出訊息
  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 用它建立連線
  ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,"tcp://127.0.0.1:61616");
  // JMS 客戶端到JMS Provider 的連線
  Connection connection = connectionFactory.createConnection();
  connection.start();
  // Session： 一個傳送或接收訊息的執行緒 true：表單開啟事務 AUTO_ACKNOWLEDGE：代表自動簽收
  Session session = connection.createSession(Boolean.TRUE,Session.AUTO_ACKNOWLEDGE);
  // Destination ：訊息的目的地;訊息傳送給誰.
  // 獲取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.自動簽收

ActiveMQ訊息簽收機制程式碼例項詳解

　　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 用它建立連線
  ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,"tcp://127.0.0.1:61616");
  // JMS 客戶端到JMS Provider 的連線
  Connection connection = connectionFactory.createConnection();
  //啟動連線
  connection.start();
  // Session： 一個傳送或接收訊息的執行緒 false:代表不帶事務的session AUTO_ACKNOWLEDGE：代表自動簽收
  /* Session session = connection.createSession(Boolean.TRUE,Session.AUTO_ACKNOWLEDGE);*/
  Session session = connection.createSession(Boolean.FALSE,Session.CLIENT_ACKNOWLEDGE);
  // Destination ：訊息的目的地;訊息傳送給誰.
  // 獲取session注意引數值my-queue是Query的名字
  Queue queue = session.createQueue("my-queue");
  // MessageProducer：建立訊息生產者
  MessageProducer producer = session.createProducer(queue);
  // 設定不持久化 PERSISTENT:代表持久化 NON_PERSISTENT：代表不持久化
  producer.setDeliveryMode(DeliveryMode.PERSISTENT);
  // 傳送訊息
  for (int i = 1; i <= 5; i++) {
   sendMsg(session,i);
  }
  System.out.println("傳送成功！");
  session.close();
  connection.close();
 }
 /**
  * 在指定的會話上，通過指定的訊息生產者發出一條訊息
  *
  * @param session
  *   訊息會話
  * @param producer
  *   訊息生產者
  */
 public static void sendMsg(Session session,int i) throws JMSException {
  // 建立一條文字訊息
  TextMessage message = session.createTextMessage("Hello ActiveMQ！" + i);
  // 通過訊息生產者發出訊息
  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 用它建立連線
  ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,"tcp://127.0.0.1:61616");
  // JMS 客戶端到JMS Provider 的連線
  Connection connection = connectionFactory.createConnection();
  connection.start();
  // Session： 一個傳送或接收訊息的執行緒 true：表單開啟事務 AUTO_ACKNOWLEDGE：代表自動簽收
  /*Session session = connection.createSession(Boolean.TRUE,Session.CLIENT_ACKNOWLEDGE);
  // Destination ：訊息的目的地;訊息傳送給誰.
  // 獲取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();
 }
}

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

相關推薦

ActiveMQ訊息簽收機制程式碼例項詳解

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

Java包裝類的快取機制原理例項詳解

這篇文章主要介紹了Java包裝類的快取機制原理例項詳解,文中通過示例程式碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下

JavaWeb Refresh響應頭程式碼例項詳解

這篇文章主要介紹了JavaWeb Refresh響應頭程式碼例項詳解,文中通過示例程式碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下

python字串,元組,列表,字典互轉程式碼例項詳解

python字串,元組,列表,字典互相轉換直接給大家上程式碼例項 #-*-coding:utf-8-*- #1、字典

JAVA類變數及類方法程式碼例項詳解

這篇文章主要介紹了JAVA類變數及類方法程式碼例項詳解,文中通過示例程式碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下

Spring RabbitMQ死信機制原理例項詳解

死信佇列：沒有被及時消費的訊息存放的佇列，訊息沒有被及時消費有以下幾點原因：

序列化模組json程式碼例項詳解

json 模組 json 是一種通用的資料傳輸格式，本質就是一個字串 json 要求 {\'key\':value}

Django呼叫支付寶介面程式碼例項詳解

支付寶支付 正式環境:用營業執照,申請商戶號,appid 測試環境:沙箱環境:https://openhome.alipay.com/platform/appDaily.htm?tab=info

Kafka Java Producer程式碼例項詳解

根據業務需要可以使用Kafka提供的Java Producer API進行產生資料，並將產生的資料傳送到Kafka對應Topic的對應分割槽中，入口類為：Producer

JavaScript array常用方法程式碼例項詳解

1.map/reduce map() map()是array的一個方法 作用: 對array中每一個元素呼叫自定義函式 \'use strict\';

Android非同步訊息處理機制實現原理詳解

訊息處理機制主要物件:Looper,Handler,Message(還有MessageQueue和Runnable) Looper不斷從MessageQueue訊息佇列中取出一個Message，然後傳給Handle，如此迴圈往復，如果佇列為空，那麼它會進入休眠。

Python爬取微信小程式通用方法程式碼例項詳解

背景介紹 最近遇到一個需求，大致就是要獲取某個小程式上的資料。心想小程式本質上就是移動端加殼的瀏覽器，所以想到用Python去獲取資料。在網上學習了一下如何實現後，記錄一下我的實現過程以及所踩過的小坑。本文關

PHP的過載使用魔術方法程式碼例項詳解

摘錄php官網對PHP過載的解釋： PHP所提供的\"過載\"（overloading）是指動態地\"建立\"類屬性和方法。我們是通過魔術方法（magic methods）來實現的。

php自動載入程式碼例項詳解

1. 讓我們為www.cppcns.com php 建立列舉，提供一些程式碼示例 如果我們的程式碼需要對列舉常量和值進行更多驗證，該怎麼辦？

PHP佇列場景以及實現程式碼例項詳解

為了降低單點壓力，通常會根據業務情況進行分表分庫，將表分佈在不同的庫中（庫可能分佈在不同的機器上）,但是一個業務場景可能會同時處理兩個表的操作。在這種場景下，事務的提交會變得相對複雜，因為多個節點（庫）

php命令列模式程式碼例項詳解

php全集行模式，即php-cli，官方文件中稱為: CLI SAPI（Server Application Programming Interface，服務端應用程式設計埠）．聽著挺複雜。其實是因為php原本為伺服器端的指令碼語言，所以引申出這個叫法。

laravel下trait的使用程式碼例項詳解

前言 今天在整理laravel的練習專案時，發現自己的程式碼結構中有很多重複的程式碼。於是搜尋了一下laravel框架的程式碼複用機制。知道了Trait的存在，於是學習使用了一下。

通過原始碼，例項詳解java類載入機制

之前的文章中，介紹了class的位元組碼靜態結構，這些類需要jvm載入到其在記憶體中分配的執行時資料區才會生效，這個過程包含：載入 -> 連結 -> 初始化 幾個階段，其中連結階段又有驗證 -> 準備 -> 解析三

wxpython多執行緒防假死與執行緒間傳遞訊息例項詳解

wxpython中啟用執行緒的方法，將GUI和功能的執行分開。 網上關於python多執行緒防假死與執行緒傳遞訊息是幾年前的，這裡由於wxpython和threading模組已經更新最新，因此給出最新修改程式碼，能在2017年最新版的pytho

Python高階程式設計之訊息佇列(Queue)與程序池(Pool)例項詳解

本文例項講述了Python高階程式設計之訊息佇列(Queue)與程序池(Pool)。分享給大家供大家參考，具體如下：