快捷導(dǎo)航

ActiveMQ消息簽收機(jī)制代碼實(shí)例詳解

更新時(shí)間：2020年02月21日 14:41:49 作者：流氓大隊(duì)長(zhǎng)

這篇文章主要介紹了ActiveMQ消息簽收機(jī)制代碼實(shí)例解析,文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下

消費(fèi)者客戶(hù)端成功接收一條消息的標(biāo)志是：這條消息被簽收。

消費(fèi)者客戶(hù)端成功接收一條消息一般包括三個(gè)階段：

1、消費(fèi)者接收消息，也即從MessageConsumer的receive方法返回

2、消費(fèi)者處理消息

3、消息被簽收

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

在帶事務(wù)的Session中，消費(fèi)者客戶(hù)端事務(wù)提交之時(shí)，消息自動(dòng)完成簽收。

在不帶事務(wù)的Session中，消息何時(shí)以及如何被簽收取決于Session的簽收模式設(shè)置

非事務(wù)Session可以設(shè)置如下幾種簽收模式：

1.Session.AUTO_ACKNOWLEDGE

當(dāng)消息從MessageConsumer的receive方法返回或者從MessageListener接口的onMessage方法返回時(shí)，會(huì)話自動(dòng)確認(rèn)消息簽收

2.Session.CLIENT_ACKNOWLEDGE

需要消費(fèi)者客戶(hù)端主動(dòng)調(diào)用acknowledge方法簽收消息，這種模式實(shí)在Session層面進(jìn)行簽收的，簽收一個(gè)已經(jīng)消費(fèi)的消息會(huì)自動(dòng)的簽收這個(gè)Session已消費(fèi)的所有消息：

例如一個(gè)消費(fèi)者在一個(gè)Session中消費(fèi)了5條消息，然后確認(rèn)第3條消息，所有這5條消息都會(huì)被簽收

3.Session.DUPS_OK_ACKNOWLEDGE

這種方式允許JMS不必急于確認(rèn)收到的消息,允許在收到多個(gè)消息之后一次完成確認(rèn),與Auto_AcKnowledge相比,這種確認(rèn)方式在某些情況下可能更有效,因?yàn)闆](méi)有確認(rèn),當(dāng)系統(tǒng)崩潰或者網(wǎng)絡(luò)出現(xiàn)故障的時(shí)候,消息可以被重新傳遞.

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

帶事務(wù)session的案例

　　生產(chǎn)者

　　　　必須在生產(chǎn)完數(shù)據(jù)之后手動(dòng)提交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 客戶(hù)端到JMS Provider 的連接
  Connection connection = connectionFactory.createConnection();
  //啟動(dòng)連接
  connection.start();
  // Session： 一個(gè)發(fā)送或接收消息的線程 false:代表不帶事務(wù)的session AUTO_ACKNOWLEDGE：代表自動(dòng)簽收
  Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
  // Destination ：消息的目的地;消息發(fā)送給誰(shuí).
  // 獲取session注意參數(shù)值my-queue是Query的名字
  Queue queue = session.createQueue("my-queue");
  // MessageProducer：創(chuàng)建消息生產(chǎn)者
  MessageProducer producer = session.createProducer(queue);
  // 設(shè)置不持久化 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();
 }
 /**
  * 在指定的會(huì)話上，通過(guò)指定的消息生產(chǎn)者發(fā)出一條消息
  *
  * @param session
  *   消息會(huì)話
  * @param producer
  *   消息生產(chǎn)者
  */
 public static void sendMsg(Session session, MessageProducer producer, int i) throws JMSException {
  // 創(chuàng)建一條文本消息
  TextMessage message = session.createTextMessage("Hello ActiveMQ！" + i);
  // 通過(guò)消息生產(chǎn)者發(fā)出消息
  producer.send(message);
 }
}

　　消費(fèi)者

　　　　消費(fèi)完數(shù)據(jù)之后必須手動(dòng)提交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 客戶(hù)端到JMS Provider 的連接
  Connection connection = connectionFactory.createConnection();
  connection.start();
  // Session： 一個(gè)發(fā)送或接收消息的線程 true：表單開(kāi)啟事務(wù) AUTO_ACKNOWLEDGE：代表自動(dòng)簽收
  Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
  // Destination ：消息的目的地;消息發(fā)送給誰(shuí).
  // 獲取session注意參數(shù)值xingbo.xu-queue是一個(gè)服務(wù)器的queue，須在在ActiveMq的console配置
  Queue queue = session.createQueue("my-queue");
  // 消費(fèi)者，消息接收者
  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();
 }
}

不帶事務(wù)session的案例

　　1.自動(dòng)簽收

　　2.手動(dòng)簽收

　　　　生產(chǎn)者

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 客戶(hù)端到JMS Provider 的連接
  Connection connection = connectionFactory.createConnection();
  //啟動(dòng)連接
  connection.start();
  // Session： 一個(gè)發(fā)送或接收消息的線程 false:代表不帶事務(wù)的session AUTO_ACKNOWLEDGE：代表自動(dòng)簽收
  /* Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);*/
  Session session = connection.createSession(Boolean.FALSE, Session.CLIENT_ACKNOWLEDGE);
  // Destination ：消息的目的地;消息發(fā)送給誰(shuí).
  // 獲取session注意參數(shù)值my-queue是Query的名字
  Queue queue = session.createQueue("my-queue");
  // MessageProducer：創(chuàng)建消息生產(chǎn)者
  MessageProducer producer = session.createProducer(queue);
  // 設(shè)置不持久化 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();
 }
 /**
  * 在指定的會(huì)話上，通過(guò)指定的消息生產(chǎn)者發(fā)出一條消息
  *
  * @param session
  *   消息會(huì)話
  * @param producer
  *   消息生產(chǎn)者
  */
 public static void sendMsg(Session session, MessageProducer producer, int i) throws JMSException {
  // 創(chuàng)建一條文本消息
  TextMessage message = session.createTextMessage("Hello ActiveMQ！" + i);
  // 通過(guò)消息生產(chǎn)者發(fā)出消息
  producer.send(message);
　　　　　message.acknowledge();　　//手動(dòng)提交
　　} 
}

　　　　消費(fèi)者

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 客戶(hù)端到JMS Provider 的連接
  Connection connection = connectionFactory.createConnection();
  connection.start();
  // Session： 一個(gè)發(fā)送或接收消息的線程 true：表單開(kāi)啟事務(wù) AUTO_ACKNOWLEDGE：代表自動(dòng)簽收
  /*Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);*/
  Session session = connection.createSession(Boolean.FALSE, Session.CLIENT_ACKNOWLEDGE);
  // Destination ：消息的目的地;消息發(fā)送給誰(shuí).
  // 獲取session注意參數(shù)值xingbo.xu-queue是一個(gè)服務(wù)器的queue，須在在ActiveMq的console配置
  Queue queue = session.createQueue("my-queue");
  // 消費(fèi)者，消息接收者
  MessageConsumer consumer = session.createConsumer(queue);
  while (true) {
   //receive():獲取消息
   TextMessage message = (TextMessage) consumer.receive();
   if (null != message) {
    System.out.println("收到消息：" + message.getText());
    message.acknowledge();　　//手動(dòng)提交
   } else {
    break;
   }
  }
  //回收資源
  session.close();
  connection.close();
 }
}

以上就是本文的全部?jī)?nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。

您可能感興趣的文章: