ActiveMQ——activemq的使用java代码实例(精选)

ActiveMQ 在java中的使用,通过单例模式、工厂实现

Jms规范里的两种message传输方式Topic和Queue,两者的对比如下表():

 

 

 

 TopicQueue
概要Publish  Subscribe messaging 发布订阅消息Point-to-Point  点对点
有无状态topic数据默认不落地,是无状态的。

Queue数据默认会在mq服务器上以文件形式保存,比如Active MQ一般保存在$AMQ_HOME\data\kr-store\data下面。也可以配置成DB存储。

完整性保障并不保证publisher发布的每条数据,Subscriber都能接受到。Queue保证每条数据都能被receiver接收。
消息是否会丢失一般来说publisher发布消息到某一个topic时,只有正在监听该topic地址的sub能够接收到消息;如果没有sub在监听,该topic就丢失了。Sender发送消息到目标Queue,receiver可以异步接收这个Queue上的消息。Queue上的消息如果暂时没有receiver来取,也不会丢失。
消息发布接收策略一对多的消息发布接收策略,监听同一个topic地址的多个sub都能收到publisher发送的消息。Sub接收完通知mq服务器一对一的消息发布接收策略,一个sender发送的消息,只能有一个receiver接收。receiver接收完后,通知mq服务器已接收,mq服务器对queue里的消息采取删除或其他操作。

 

一、导jar包

 activemq的依赖包

<dependency><groupId>org.slf4j</groupId><artifactId>slf4j-log4j12</artifactId><version>4.12</version></dependency><dependency><groupId>org.slf4j</groupId><artifactId>jul-to-slf4j</artifactId><version>1.6.1</version></dependency><dependency><groupId>org.apache.activemq</groupId><artifactId>activemq-all</artifactId><version>5.13.3</version></dependency><dependency><groupId>org.springframework</groupId><artifactId>spring-jms</artifactId><version>4.3.1.RELEASE</version><exclusions><exclusion><groupId>org.springframework</groupId><artifactId>spring-messaging</artifactId></exclusion><exclusion><groupId>org.springframework</groupId><artifactId>spring-context</artifactId></exclusion><exclusion><groupId>org.springframework</groupId><artifactId>spring-beans</artifactId></exclusion><exclusion><groupId>org.springframework</groupId><artifactId>spring-aop</artifactId></exclusion><exclusion><groupId>org.springframework</groupId><artifactId>spring-tx</artifactId></exclusion><exclusion><groupId>org.springframework</groupId><artifactId>spring-core</artifactId></exclusion></exclusions></dependency>

 

 

二、java代码

 创建一下四个java文件,成为mq的公共数据连接池

1、连接工厂 配置

package com.broadsense.iov.base.jms;import org.apache.activemq.ActiveMQConnectionFactory;
import org.springframework.jms.connection.CachingConnectionFactory;
/*** 连接工厂 配置* * @author flm* 2017年10月13日*/
public class ConnectionFactory
{private static final String URL = "tcp://10.10.1.1:61616";private static final String USERNAME = "hkadmin";private static final String PASSWORD = "hk667";private static final int SESSIONCACHESIZE = 20;private javax.jms.ConnectionFactory factory;public static synchronized javax.jms.ConnectionFactory getInstance(){if (SingletonHolder.INSTANCE.factory == null) {SingletonHolder.INSTANCE.build();}return SingletonHolder.INSTANCE.factory;}private void build(){AMQConfigBean bean = loadConfigure();this.factory = buildConnectionFactory(bean);}private javax.jms.ConnectionFactory buildConnectionFactory(AMQConfigBean bean) {javax.jms.ConnectionFactory targetFactory = new ActiveMQConnectionFactory(bean.getUserName(), bean.getPassword(), bean.getBrokerURL());CachingConnectionFactory connectoryFacotry = new CachingConnectionFactory();connectoryFacotry.setTargetConnectionFactory(targetFactory);connectoryFacotry.setSessionCacheSize(bean.getSessionCacheSize());return connectoryFacotry;}private AMQConfigBean loadConfigure() {if ("tcp://10.10.1.1:61616" != null) {try {return new AMQConfigBean("tcp://10.10.1.1:61616", "hkadmin", "hk667", 20);} catch (Exception e) {throw new IllegalStateException("load amq config error!");}}throw new IllegalStateException("load amq config error!");}private static class AMQConfigBean{private String brokerURL;private String userName;private String password;private int sessionCacheSize;public AMQConfigBean() {}public AMQConfigBean(String brokerURL, String userName, String password, int sessionCacheSize) {this.brokerURL = brokerURL;this.userName = userName;this.password = password;this.sessionCacheSize = sessionCacheSize;}public String getBrokerURL() {return this.brokerURL;}public void setBrokerURL(String brokerURL) {this.brokerURL = brokerURL;}public String getUserName() {return this.userName;}public void setUserName(String userName) {this.userName = userName;}public String getPassword() {return this.password;}public void setPassword(String password) {this.password = password;}public int getSessionCacheSize() {return this.sessionCacheSize;}public void setSessionCacheSize(int sessionCacheSize) {this.sessionCacheSize = sessionCacheSize;}}private static class SingletonHolder{static ConnectionFactory INSTANCE = new ConnectionFactory(null);}
}

 

2、模版

package com.broadsense.iov.base.jms;import org.springframework.jms.core.JmsTemplate;

/**
* 模板厂
*
* @author flm
* 2017年10月13日
*/

public class JmsTemplateFactory
{private final javax.jms.ConnectionFactory factory;private JmsTemplate topicJmsTemplate;private JmsTemplate queueJmsTemplate;public static JmsTemplateFactory getInstance(){return SingletonHolder.INSTANCE;}private JmsTemplateFactory(){this.factory = ConnectionFactory.getInstance();}public synchronized JmsTemplate getTopicJmsTemplate() {if (this.topicJmsTemplate == null) {this.topicJmsTemplate = createTemplate(this.factory, true);}return this.topicJmsTemplate;}public synchronized JmsTemplate getQueueJmsTemplate() {if (this.queueJmsTemplate == null) {this.queueJmsTemplate = createTemplate(this.factory, false);}return this.queueJmsTemplate;}private JmsTemplate createTemplate(javax.jms.ConnectionFactory factory, boolean pubSubDomain) {JmsTemplate template = new JmsTemplate(factory);template.setPubSubDomain(pubSubDomain);return template;}public static class SingletonHolder{static JmsTemplateFactory INSTANCE = new JmsTemplateFactory(null);}
}

 

3、消费者 模版

package com.broadsense.iov.base.jms;import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import javax.jms.Destination;
import javax.jms.MessageListener;
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.command.ActiveMQTopic;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jms.listener.SimpleMessageListenerContainer;
/*** JMS监听器  创建消费者* * @author flm* 2017年10月13日*/
public class JMSListener
{private static final Logger LOGGER = LoggerFactory.getLogger(JMSListener.class);private static final Map<String, Destination> MQDESTS = new ConcurrentHashMap();


  

  /**
  * 开启一个 点对点的 消息队列监听 的消费者
  *
  * @param queueName 队列名称
  * @param subName 订阅者的名字
  * @param listener 监听
  */

   public static synchronized void startJmsQueueListener(String queueName, MessageListener listener)

  {startJmsQueueListener(queueName, null, listener);}public static synchronized void startJmsQueueListener(String queueName, String subName, MessageListener listener) {Destination dst = (Destination)MQDESTS.get("QUEUE_" + queueName);if (dst == null) {ActiveMQQueue mq = new ActiveMQQueue(queueName);startJmsListener(mq, subName, listener);MQDESTS.put("QUEUE_" + queueName, mq);} else {LOGGER.warn(queueName + " already started");}}


  /**
  * 开启 一对多 主题的 消息监听的消费者
  *
  * @param topicName 主题消息名称
  * @param subName 订阅者的名字
  * @param listener 监听
  */

public static synchronized void startJmsTopicListener(String topicName, MessageListener listener){startJmsTopicListener(topicName, null, listener);}public static synchronized void startJmsTopicListener(String topicName, String subName, MessageListener listener) {ActiveMQTopic mq = new ActiveMQTopic(topicName);startJmsListener(mq, subName, listener);MQDESTS.put("QUEUE_" + topicName, mq);}

  

  /**
  * 开始 消息监听器 消费者
  *
  * @param dest 目的地
  * @param subName 持久订阅的名字
  * @param msgListener 消息监听器
  */

private static void startJmsListener(Destination dest, String subName, MessageListener msgListener){javax.jms.ConnectionFactory factory = ConnectionFactory.getInstance();SimpleMessageListenerContainer listener = new SimpleMessageListenerContainer();listener.setConnectionFactory(factory);listener.setDestination(dest);listener.setMessageListener(msgListener);if ((subName != null) && (subName != "")) {listener.setDurableSubscriptionName(subName);}listener.start();}
}

 

4、生产者 模版

package com.broadsense.iov.base.jms;import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;/*** 创建 jms生产者* * @author flm* 2017年10月13日*/
public class JMSPublisher
{

  


  /**
  * 发送消息
  * Topic 生产者
  *
  * @param dest 目的地
  * @param msg 消息内容
  */

public static void sendTopicMessage(String dest, String msg){JmsTemplateFactory.getInstance().getTopicJmsTemplate().send(dest, new MessageCreator(msg){public Message createMessage(Session session) throws JMSException {return session.createTextMessage(this.val$msg);}});}


  /**
  * 发送消息
  * Queue 生产者
  *
  * @param dest 目的地
  * @param msg 消息内容
  */

public static void sendQueueMessage(String dest, String msg){JmsTemplateFactory.getInstance().getQueueJmsTemplate().send(dest, new MessageCreator(msg){public Message createMessage(Session session) throws JMSException {return session.createTextMessage(this.val$msg);}});}
}

 

 

三、activemq的使用

1、创建一个junit测试,@Test 发布、接受、即可看到消息,mq管理后台也可以看到

package com.broadsense.iov.base.jms;import com.broadsense.iov.base.jms.JMSListener;
import com.broadsense.iov.base.jms.JMSPublisher;
import java.io.IOException;
import java.util.logging.Level;
import java.util.logging.Logger;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;
import org.junit.Test;/**** @author flm*/
public class JMSPublisherTest {public JMSPublisherTest() {}/*** 生产者 发布消息* @throws */@Testpublic void testSendMessage() throws InterruptedException {for (int idx = 1; idx < 3; idx++) {/** 生产者 发布 消息到 queue/queue_b 的队列中*/JMSPublisher.sendQueueMessage("queue/queue_b", String.valueOf(idx * 1111));/** 生产者 发布消息 到  topic/send 的Topic 主题中 *///JMSPublisher.sendTopicMessage("topic/send", String.valueOf(idx * 1111));
        }}/*** 消费者 订阅接受消息*/@Testpublic void receiver() {/** 消费者 订阅主题  topic/send 是否有消息发布,有侧打印出来  (通过 onMessage 监听)*//*JMSListener.startJmsTopicListener("topic/send", new MessageListener() {@Overridepublic void onMessage(Message message) {try {if (message instanceof TextMessage) {TextMessage msg = (TextMessage) message;System.out.println("== 收到一个JMS消息..." + msg.getText());} } catch (JMSException e) {e.printStackTrace();}}});*//** 消费者 订阅队列  queue/queue_b 是否有消息发布,有侧打印出来  (通过 onMessage 监听)*/JMSListener.startJmsQueueListener("queue/queue_b" ,new MessageListener() {@Overridepublic void onMessage(Message message) {try {if (message instanceof TextMessage) {TextMessage msg = (TextMessage) message;System.out.println("== 收到一个JMS消息..." + msg.getText());} } catch (JMSException e) {e.printStackTrace();}}});try {System.in.read();} catch (IOException ex) {Logger.getLogger(JMSPublisherTest.class.getName()).log(Level.SEVERE, null, ex);}}}
View Code

 

 2、真正的项目实现

在项目的中具体实现,是加载一个类来实现订阅消息

加载启动一个订阅的主题,给一个类MQ()处理

package com.ifengSearch.track.dao;import org.springframework.stereotype.Repository;import com.broadsense.iov.base.jms.JMSListener;/*** 项目启动即 开启* 通过 spring 依赖加载 Lister 订阅topic/send* @author flm* @2017年10月16日*/
@Repository
public class Lister {public Lister(){try {JMSListener.startJmsTopicListener("topic/send",new QM());// QM() 订阅 主题  topic/send} catch (Exception e) {}}
}
View Code

 

MQ()订阅消息的处理类,通过实现

package com.ifengSearch.track.dao;import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;/*** 通过 实现 MessageListener 的 onMessage 来监听消息 * 接受、处理消息* @author flm* @2017年10月16日*/
public class MQ implements MessageListener {@Overridepublic void onMessage(Message message) {try {if (message instanceof TextMessage) {TextMessage msg = (TextMessage) message;System.out.println("== 收到一个JMS消息..." + msg.getText());} } catch (JMSException e) {e.printStackTrace();}}
}
View Code

 

转载于:https://www.cnblogs.com/lemon-flm/p/7668076.html

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/news/368979.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

Java 8:正在运行的CompletableFuture

在Java 8中全面研究了CompletableFuture API之后&#xff0c;我们准备编写一个简单的Web搜寻器。 我们已经使用ExecutorCompletionService &#xff0c; Guava ListenableFuture和Scala / Akka解决了类似的问题。 我选择了相同的问题&#xff0c;以便轻松比较方法和实现技术。 …

java能不能修改文件大小信息_java上机考试3

考号: 000128000091 姓名: 张天義 学号: 1040610329 班级: 本科(计算机、微电、电工、经管)->计算机科学与技术 00:13:25一、单选(共80分)1. int count1;for(int i1;i<5;i){countcounti;}System.out.println(count);上述代码执行后的输出结果是A、15B、1C、16D、5标记此题…

(网页)中的简单的遮罩层

html: <div id"test"></div><div id"log_window"> <!--<a href"javascript:cancel_shield()">关闭</a>--> <!--<img src"assets/css/images/loading.gif" />--> <p style"fo…

即使在jdk中也有错误的代码

Java 7&#xff0c;TreeSet和NullPointerException。 最近&#xff0c;我尝试用Java 7编译一个用Java 6开发的项目。在执行测试过程中发生了很多有趣的事情&#xff0c;在Java 6中使用Java 7平稳运行的测试失败了&#xff01; 因此&#xff0c;我必须理解为什么&#xff0c;这就…

Eclipse+GitHub 提交代码错误 -“rejected - non-fast-forward”

Eclipse Push出现rejected - non-fast-forward错误 在 Push到服务器时有时会出现 rejected - non-fast-forward 错误&#xff0c;这是由于远端发生改变&#xff0c;此时再提交之前你需要将远端的改变合并到本地上错误原因&#xff1a;文件冲突&#xff0c;本地的代码和远程Repo…

两天学会css基础(一)

什么是css&#xff1f;css的作用是什么&#xff1f; CSS 指层叠样式表 (Cascading Style Sheets)主要作用就是给HTML结构添加样式&#xff0c;搭建页面结构&#xff0c;比如设置元素的宽高大小&#xff0c;颜色&#xff0c;位置等等。 学习css之前先了解一下css代码在HTML中的…

在Android项目中使用AspectJ

版权声明&#xff1a;本文为博主原创文章&#xff0c;未经博主允许不得转载。 转载请表明出处&#xff1a;http://www.cnblogs.com/cavalier-/p/8888459.html 什么是AOP AOP是 Aspect Oriented Programming 的缩写&#xff0c;即面向切面编程&#xff0c;和平常遇到的面向对象O…

LVM 逻辑卷 (logica volume manager)

逻辑卷轴管理员 (Logical Volume Manager) 想像一个情况&#xff0c;你在当初规划主机的时候将 /home 只给他 50G &#xff0c;等到使用者众多之后导致这个 filesystem 不够大&#xff0c; 此时你能怎么作&#xff1f; 多数的朋友都是这样&#xff1a;再加一颗新硬盘&#xff0…

java中u怎么用_Java中interrupt的使用

通常我们会有这样的需求&#xff0c;即停止一个线程。在java的api中有stop、suspend等方法可以达到目的&#xff0c;但由于这些方法在使用上存在不安全性&#xff0c;会带来不好的副作用&#xff0c;不建议被使用。具体原因可以参考Why is Thread.stop deprecated。在本文中&am…

当Maven依赖插件位于

问题&#xff1a; 我们进行了一个集成测试&#xff0c;该测试创建了一个Spring ClassPathXmlApplicationContext &#xff0c;同时这样做导致NoSuchMethodError爆炸。 事实证明&#xff0c;我们对Spring构件的依赖版本存在冲突。 这本身不是一个不寻常的问题-使用Maven依赖插件…

sql查询语句for xml path语法

【原地址】 for xml path作用&#xff1a;将多行的查询结果&#xff0c;根据某一些条件合并到一行。 例&#xff1a;现有一张表 执行下面语句 select Department,(SELECT Employee, FROM People b WHERE b.Departmenta.Department For XML Path()) Student from People as a g…

css高度已知,左右定宽,中间自适应三栏布局

css高度已知&#xff0c;左右定宽&#xff0c;中间自适应三栏布局&#xff1a; <!DOCTYPE html> <html lang"en"> <head><meta charset"UTF-8"><meta name"viewport" content"widthdevice-width, initial-scale…

java使用impala存放多条sql_Impala基于内存的SQL引擎的详细介绍

数据存储使用相同的存储数据池都支持把数据存储于HDFS, HBase。元数据&#xff1a;两者使用相同的元数据SQL解释处理&#xff1a;比较相似都是通过词法分析生成执行计划。执行计划&#xff1a;Hive: 依赖于MapReduce执行框架&#xff0c;执行计划分成 map->shuffle->redu…

Android Studio打包以及Gradle配置构建

本文转载 郭霖公众号 https://mp.weixin.qq.com/s?__bizMzA5MzI3NjE2MA&mid2650241610&idx1&snb8af73f6c288b6617d9fe0ab3618118d&pass_ticketQK4j37kpmGNlsYcECWMb64HxKHEVJG5mSJubQEQguKI%3D 生成签名文件手动打包 首先生成签名文件&#xff0c;点击 Build…

去除inline-block间隙的几种方法

为什么会产生间隙&#xff1f; 由于编写代码时的美观和可读性&#xff0c;在代码中添加回车或空格而产生的间隙。 html代码&#xff1a; <ul class"container"><li></li><li></li><li></li><li></li><li&…

java重载方法math_Java语言程序设计(十二)Math数学类,方法重载及变量作用域...

1.重载方法上一篇文章用到的max方法只能用于int型数据类型&#xff0c;但是如果需要决定两个浮点数中哪个较大&#xff0c;解决方法是创建另一个方法名相同但参数不同的方法&#xff0c;代码如下&#xff1a;public static double max(double num1, double num2){if(num1>nu…

编码(转)

https://www.zhihu.com/question/28164512 关于编码和乱码的问题&#xff0c;我简单讲一下。 通常问这类问题的人是混淆了若干个不同的概念&#xff0c;并且他们自己也没有意识到自己混淆了这些概念的。 终端显示字符的编码&#xff08;windows下终端是cmd&#xff0c;linux下是…

Spring MVC:测试简介

测试是软件开发中最重要的部分之一。 井井有条的测试有助于使应用程序代码保持良好状态&#xff0c;并且处于工作状态。 有很多不同类型的测试和方法。 在本文中&#xff0c;我想对基于Spring MVC的应用程序进行单元测试进行介绍。 不要希望在这里阅读有关Spring MVC测试的全部…

yaml,json,ini这三种格式用来做配置文件优缺点

适合人类编写&#xff1a;ini > toml > yaml > json > xml > plist可以存储的数据复杂度&#xff1a;xml > yaml > toml ~ json ~ plist > ini 作者&#xff1a;赵扶摇链接&#xff1a;https://www.zhihu.com/question/41253282/answer/119857880来源&…

试验ConcurrentHashmap

我正在研究我最近的一个项目中的内存问题&#xff0c;该项目将数据保留在内存中以进行快速访问&#xff0c;但是应用程序的内存占用量非常大。 该应用程序大量使用CHM&#xff08;即Concurrenthashmap&#xff09; &#xff0c;因此&#xff0c;无需再费脑筋地猜测CHM是问题所…