Java消息队列--ActiveMq 初体验

1、下载安装ActiveMQActiveMQ官网下载地址：http://activemq.apache.org/download.htmlActiveMQ 提供了Windows 和Linux、Unix 等几个版本，楼主这里选择了Linux 版本下进行开发。从它的目录来说，还是很简单的：
1.bin存放的是脚本文件
2.conf存放的是基本配置文件
3.data存放的是日志文件
4.docs存放的是说明文档
5.examples存放的是简单的实例
6.lib存放的是activemq所需jar包
7.webapps用于存放项目的目录2、启动ActiveMQ　
进入到ActiveMQ 安装目录的Bin 目录，linux 下输入 ./activemq start 启动activeMQ 服务。
输入命令之后，会提示我们创建了一个进程IP 号，这时候说明服务已经成功启动了。
ActiveMQ默认启动时，启动了内置的jetty服务器，提供一个用于监控ActiveMQ的admin应用。 admin：http://127.0.0.1:8161/admin/
我们在浏览器打开链接之后输入账号密码（这里和tomcat 服务器类似）
默认账号：admin
密码：admin到这里为止，ActiveMQ 服务端就启动完毕了。
ActiveMQ 在linux 下的终止命令是 ./activemq stop3、创建一个ActiveMQ工程
上述在官网下载ActiveMq 的时候，我们可以在目录下看到一个jar包：
这个jar 包就是我们需要在项目中进行开发中使用到的相关依赖。3.1 创建生产者
public class Producter {//ActiveMq 的默认用户名private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;//ActiveMq 的默认登录密码private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;//ActiveMQ 的链接地址private static final String BROKEN_URL = ActiveMQConnection.DEFAULT_BROKER_URL;AtomicInteger count = new AtomicInteger(0);//链接工厂ConnectionFactory connectionFactory;//链接对象Connection connection;//事务管理Session session;ThreadLocal<MessageProducer> threadLocal = new ThreadLocal<>();public void init(){try {//创建一个链接工厂connectionFactory = new ActiveMQConnectionFactory(USERNAME,PASSWORD,BROKEN_URL);//从工厂中创建一个链接connection  = connectionFactory.createConnection();//开启链接connection.start();//创建一个事务（这里通过参数可以设置事务的级别）session = connection.createSession(true,Session.SESSION_TRANSACTED);} catch (JMSException e) {e.printStackTrace();}}public void sendMessage(String disname){try {//创建一个消息队列Queue queue = session.createQueue(disname);//消息生产者MessageProducer messageProducer = null;if(threadLocal.get()!=null){messageProducer = threadLocal.get();}else{messageProducer = session.createProducer(queue);threadLocal.set(messageProducer);}while(true){Thread.sleep(1000);int num = count.getAndIncrement();//创建一条消息TextMessage msg = session.createTextMessage(Thread.currentThread().getName()+"productor:我是大帅哥，我现在正在生产东西！,count:"+num);System.out.println(Thread.currentThread().getName()+"productor:我是大帅哥，我现在正在生产东西！,count:"+num);//发送消息messageProducer.send(msg);//提交事务session.commit();}} catch (JMSException e) {e.printStackTrace();} catch (InterruptedException e) {e.printStackTrace();}}
}public class Comsumer {private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;private static final String BROKEN_URL = ActiveMQConnection.DEFAULT_BROKER_URL;ConnectionFactory connectionFactory;Connection connection;Session session;ThreadLocal<MessageConsumer> threadLocal = new ThreadLocal<>();AtomicInteger count = new AtomicInteger();public void init(){try {connectionFactory = new ActiveMQConnectionFactory(USERNAME,PASSWORD,BROKEN_URL);connection  = connectionFactory.createConnection();connection.start();session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);} catch (JMSException e) {e.printStackTrace();}}public void getMessage(String disname){try {Queue queue = session.createQueue(disname);MessageConsumer consumer = null;if(threadLocal.get()!=null){consumer = threadLocal.get();}else{consumer = session.createConsumer(queue);threadLocal.set(consumer);}while(true){Thread.sleep(1000);TextMessage msg = (TextMessage) consumer.receive();if(msg!=null) {msg.acknowledge();System.out.println(Thread.currentThread().getName()+
": Consumer:我是消费者，我正在消费Msg"+msg.getText()+"--->"+count.getAndIncrement());}else {break;}}} catch (JMSException e) {e.printStackTrace();} catch (InterruptedException e) {e.printStackTrace();}}
}4、运行ActiveMQ项目
4.1 生产者开始生产消息
public class TestMq {public static void main(String[] args){Producter producter = new Producter();producter.init();TestMq testMq = new TestMq();try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}//Thread 1new Thread(testMq.new ProductorMq(producter)).start();//Thread 2new Thread(testMq.new ProductorMq(producter)).start();//Thread 3new Thread(testMq.new ProductorMq(producter)).start();//Thread 4new Thread(testMq.new ProductorMq(producter)).start();//Thread 5new Thread(testMq.new ProductorMq(producter)).start();}private class ProductorMq implements Runnable{Producter producter;public ProductorMq(Producter producter){this.producter = producter;}@Overridepublic void run() {while(true){try {producter.sendMessage("Jaycekon-MQ");Thread.sleep(10000);} catch (InterruptedException e) {e.printStackTrace();}}}}
}

ActiveMQ详细入门使用教程

ActiveMQ介绍MQ是消息中间件，是一种在分布式系统中应用程序借以传递消息的媒介，常用的有ActiveMQ，
RabbitMQ，kafka。ActiveMQ是Apache下的开源项目，完全支持JMS1.1和J2EE1.4规范的JMS Provider实现。 特点：
1、支持多种语言编写客户端
2、对spring的支持，很容易和spring整合
3、支持多种传输协议：TCP,SSL,NIO,UDP等
4、支持AJAX 消息形式：
1、点对点（queue）
2、一对多（topic） ActiveMQ测试编写一个测试类对ActiveMQ进行测试，首先得向pom文件中添加ActiveMQ相关的jar包：<dependency>  <groupId>org.apache.activemq</groupId>  <artifactId>activemq-all</artifactId>  </dependency>  queue的发送代码如下:public void testMQProducerQueue() throws Exception{//1、创建工厂连接对象，需要制定ip和端口号ConnectionFactory connectionFactory
= new ActiveMQConnectionFactory("tcp://192.168.156.44:61616");//2、使用连接工厂创建一个连接对象Connection connection = connectionFactory.createConnection();//3、开启连接connection.start();//4、使用连接对象创建会话（session）对象Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);//5、使用会话对象创建目标对象，包含queue和topic（一对一和一对多）Queue queue = session.createQueue("test-queue");//6、使用会话对象创建生产者对象MessageProducer producer = session.createProducer(queue);//7、使用会话对象创建一个消息对象TextMessage textMessage = session.createTextMessage("hello!test-queue");//8、发送消息producer.send(textMessage);//9、关闭资源producer.close();session.close();connection.close();}接收代码：public void TestMQConsumerQueue() throws Exception{//1、创建工厂连接对象，需要制定ip和端口号ConnectionFactory connectionFactory
= new ActiveMQConnectionFactory("tcp://192.168.156.44:61616");//2、使用连接工厂创建一个连接对象Connection connection = connectionFactory.createConnection();//3、开启连接connection.start();//4、使用连接对象创建会话（session）对象Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);//5、使用会话对象创建目标对象，包含queue和topic（一对一和一对多）Queue queue = session.createQueue("test-queue");//6、使用会话对象创建生产者对象MessageConsumer consumer = session.createConsumer(queue);//7、向consumer对象中设置一个messageListener对象，用来接收消息consumer.setMessageListener(new MessageListener() {@Overridepublic void onMessage(Message message) {// TODO Auto-generated method stubif(message instanceof TextMessage){TextMessage textMessage = (TextMessage)message;try {System.out.println(textMessage.getText());} catch (JMSException e) {// TODO Auto-generated catch blocke.printStackTrace();}}}});//8、程序等待接收用户消息System.in.read();//9、关闭资源consumer.close();session.close();connection.close();}接着对topic进行测试，发送代码如下：public void TestTopicProducer() throws Exception{//1、创建工厂连接对象，需要制定ip和端口号ConnectionFactory connectionFactory =
new ActiveMQConnectionFactory("tcp://192.168.156.44:61616");//2、使用连接工厂创建一个连接对象Connection connection = connectionFactory.createConnection();//3、开启连接connection.start();//4、使用连接对象创建会话（session）对象Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);//5、使用会话对象创建目标对象，包含queue和topic（一对一和一对多）Topic topic = session.createTopic("test-topic");//6、使用会话对象创建生产者对象MessageProducer producer = session.createProducer(topic);//7、使用会话对象创建一个消息对象TextMessage textMessage = session.createTextMessage("hello!test-topic");//8、发送消息producer.send(textMessage);//9、关闭资源producer.close();session.close();connection.close();}接收代码：
然后运行topic发送：
可以看到消息已经发送出去。再运行topic接收：ActiveMQ整合spring及项目中运用
activeMQ与spring看一整合到一起使用，除了添加ActiveMQ相关的jar包外，还需要添加spring的jar包：<dependency>  <groupId>org.springframework</groupId>  <artifactId>spring-context</artifactId>  </dependency>  <!-- 配置能够产生connection的connectionfactory，由JMS对应的服务厂商提供 --><bean id="tagertConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory"><constructor-arg name="brokerURL" value="tcp://192.168.156.44:61616"/></bean><!-- 配置spring管理真正connectionfactory的connectionfactory，相当于spring
对connectionfactory的一层封装 --><bean
id="connectionFactory" lass="org.springframework.jms.connection.
SingleConnectionFactory"><property name="targetConnectionFactory" ref="tagertConnectionFactory"/></bean><!-- 配置生产者 --><!-- Spring使用JMS工具类，可以用来发送和接收消息 --><bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate"><!-- 这里是配置的spring用来管理connectionfactory的connectionfactory --><property name="connectionFactory" ref="connectionFactory"/></bean><!-- 配置destination --><!-- 队列目的地 --><bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue"><constructor-arg value="spring-queue"/></bean><!-- 话题目的地 --><bean id="itemAddTopic" class="org.apache.activemq.command.ActiveMQTopic"><constructor-arg value="item-add-topic"/></bean>当然，在xml文件中配置好的jmstemplate和destination也要注入进来：@Autowiredprivate JmsTemplate jmsTemplate;@Resource(name="itemAddTopic")private Destination destination;然后消费者应该写在我们的搜索工程中，首先添加spring和activeMQ的jar包，然后配置xml文件，
再编写一个监听器，当接收到消息时，就讲数据存入索引库，xml文件代码如下：<!-- 配置监听器 --><bean id="myListener" class="com.taotao.search.listener.MyListener"/><bean id="itemAddListener" class="com.taotao.search.listener.ItemAddListener"/><!-- 系统监听器 -->
<!--    <bean class="org.springframework.jms.listener.DefaultMessageListenerContainer"><property name="connectionFactory" ref="connectionFactory"/><property name="destination" ref="queueDestination"/><property name="messageListener" ref="myListener"/></bean> --><bean class="org.springframework.jms.listener.DefaultMessageListenerContainer"><property name="connectionFactory" ref="connectionFactory"/><property name="destination" ref="itemAddTopic"/><property name="messageListener" ref="itemAddListener"/>

Java消息队列--ActiveMq 初体验相关推荐

java消息队列ActiveMQ之Comsumer
1.需要添加在上一篇博客中说到的依赖jar包 2.创建创建消费者 package com.activemq.demo;import java.util.concurrent.atomic.Atomic ...
Java消息队列-Spring整合ActiveMq
1.概述首先和大家一起回顾一下Java 消息服务,在我之前的博客<Java消息队列-JMS概述>中,我为大家分析了: 消息服务:一个中间件,用于解决两个或多个程序之间的耦合,底层由Jav ...
java 消息队列详解_Java消息队列-Spring整合ActiveMq的详解
本篇文章主要介绍了详解Java消息队列-Spring整合ActiveMq ,小编觉得挺不错的,现在分享给大家,也给大家做个参考.一起跟随小编过来看看吧 1.概述首先和大家一起回顾一下Java 消息服 ...
java分布式面试题之消息队列ActiveMQ部分
java分布式面试题之消息队列ActiveMQ部分 java分布式面试题之消息队列ActiveMQ部分 1.如何使用ActiveMQ解决分布式事务? 在互联网应用中,基本都会有用户注册的功能.在注册的 ...
java 消息队列_java消息队列
Java消息服务(Java Message Service,JMS)应用程序接口是一个Java平台中关于面向消息中间件(MOM)的API,用于在两个应用程序之间,或分布式系统中发送消息,进行异步通信. ...
CentOS源码安装消息队列ActiveMQ
消息队列ActiveMQ介绍 JMS全称:Java Message Service中文:Java消息服务.JMS是java的一套API标准,最初的目的是为了是应用程序能够访问现有的MOM系统(MOM是 ...
Java 消息队列、缓存、同步（个人理解：空谈）
Java 消息队列.缓存.同步. 消息队列我的理解:消息队列,将消息存入消息队列,然后就OK了. 系统之间原先调用通用接口,但引入了消息队列后,系统之间调用MQ消息队列. 好处:响应快,能累积请求, ...
消息队列 - ActiveMQ
消息队列 - ActiveMQ 一.入门概述 1.在什么场景下使用消息中间件,为什么使用 2. 消息队列是什么 3. 去哪下二.ActiveMQ的安装和控制台 1.Linux安装 2.Apache ...
Java开发 - 单点登录初体验（Spring Security + JWT）
目录前言为什么要登录登录的种类 Cookie-Session Cookie-Session-local storage JWT令牌几种登陆总结用户身份认证与授权创建工程添加 ...

Java消息队列--ActiveMq 初体验

Java消息队列--ActiveMq 初体验相关推荐

最新文章

热门文章