ActiveMQ消息签收机制代码实例详解

 更新时间:2020年02月21日 14:41:49   作者:流氓大队长  
这篇文章主要介绍了ActiveMQ消息签收机制代码实例解析,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下

这篇文章主要介绍了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,
    ActiveMQConnection.DEFAULT_PASSWORD, "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.自动签收

  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,
    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);*/
  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, producer, i);
  }
  System.out.println("发送成功!");
  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);
     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,
    ActiveMQConnection.DEFAULT_PASSWORD, "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);*/
  Session session = connection.createSession(Boolean.FALSE, 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();
 }
}

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持脚本之家。

相关文章

  • java微信企业号开发之开发模式的开启

    java微信企业号开发之开发模式的开启

    这篇文章主要为大家详细介绍了java微信企业号开发之开发模式的开启方法,感兴趣的小伙伴们可以参考一下
    2016-06-06
  • 使用Java将字符串在ISO-8859-1和UTF-8之间相互转换

    使用Java将字符串在ISO-8859-1和UTF-8之间相互转换

    大家都知道在一些情况下,我们需要特殊的编码格式,如:UTF-8,但是系统默认的编码为ISO-8859-1,遇到这个问题,该如何对字符串进行两个编码的转换呢,下面小编给大家分享下java中如何在ISO-8859-1和UTF-8之间相互转换,感兴趣的朋友一起看看吧
    2021-12-12
  • SpringAOP中的动态代理技术深入解析

    SpringAOP中的动态代理技术深入解析

    这篇文章主要介绍了SpringAOP中的动态代理技术深入解析,spring默认使用JDK动态代理实现AOP,类如果实现了接口,spring就会用JDK动态代理实现AOP,如果目标类没有实现接口,spring则使用Cglib动态代理来实现AOP,需要的朋友可以参考下
    2024-01-01
  • Java多线程 Callable、Future 和FutureTask

    Java多线程 Callable、Future 和FutureTask

    这篇文章主要介绍Java多线程中的 Callable、Future 以及FutureTask,下面文章围绕Java多线程的相关资料展开全文详细内容,需要的朋友可以参考一下
    2021-10-10
  • Java 流的高级使用之收集数据解析

    Java 流的高级使用之收集数据解析

    这篇文章主要介绍了Java 流的高级使用之收集数据解析,具有很好的参考价值,希望对大家有所帮助。如有错误或未考虑完全的地方,望不吝赐教
    2021-08-08
  • Mybatis如何使用ognl表达式实现动态sql

    Mybatis如何使用ognl表达式实现动态sql

    这篇文章主要介绍了Mybatis使用ognl表达式实现动态sql的操作,具有很好的参考价值,希望对大家有所帮助。如有错误或未考虑完全的地方,望不吝赐教
    2021-06-06
  • SpringBoot3使用Jasypt实现加密配置文件

    SpringBoot3使用Jasypt实现加密配置文件

    这篇文章主要为大家详细介绍了SpringBoot3如何使用Jasypt实现加密配置文件功能,文中的示例代码讲解详细,感兴趣的小伙伴可以跟随小编一起学习一下
    2025-03-03
  • Java重写equals及hashcode方法流程解析

    Java重写equals及hashcode方法流程解析

    这篇文章主要介绍了Java重写equals及hashcode方法流程解析,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
    2020-04-04
  • 解析spring boot与ireport 整合问题

    解析spring boot与ireport 整合问题

    本文通过实例代码给大家介绍了spring boot 与 ireport 整合问题,关于pom文件依赖的问题通过实例代码给大家介绍的非常详细,需要的朋友参考下吧
    2021-10-10
  • 使用jar包反编译形成pom工程

    使用jar包反编译形成pom工程

    这篇文章主要介绍了使用jar包反编译形成pom工程,具有很好的参考价值,希望对大家有所帮助。如有错误或未考虑完全的地方,望不吝赐教
    2022-06-06

最新评论