文档章节

ActiveMQ消息队列

尚浩宇
 尚浩宇
发布于 2015/08/24 14:27
字数 1549
阅读 495
收藏 3

        下载地址:官方下载地址

        下载完成后解压缩apache-activemq-5.5.1-bin.zip,然后双击apache-activemq-5.5.1\bin\activemq.bat运行ActiveMQ程序。    

wrapper  | --> Wrapper Started as Console
wrapper  | Launching a JVM...
jvm 1    | Wrapper (Version 3.2.3) http://wrapper.tanukisoftware.org
jvm 1    |   Copyright 1999-2006 Tanuki Software, Inc.  All Rights Reserved.
jvm 1    |
jvm 1    | Java Runtime: Oracle Corporation 1.7.0_51 C:\Program Files\Java\jdk1.7.0_51\jre
jvm 1    |   Heap sizes: current=61440k  free=56514k  max=932352k
jvm 1    |     JVM args: -Dactivemq.home=../.. -Dactivemq.base=../.. -Djavax.net.ssl.keyStorePassword=password -Djavax.net.ssl.trustStorePassword=password -Djavax.net.ssl.keyStore=../../conf/broker.ks
 -Djavax.net.ssl.trustStore=../../conf/broker.ts -Dcom.sun.management.jmxremote -Dorg.apache.activemq.UseDedicatedTaskRunner=true -Djava.util.logging.config.file=logging.properties -Dactivemq.conf=../
../conf -Dactivemq.data=../../data -Djava.security.auth.login.config=../../conf/login.config -Xmx1024m -Djava.library.path=../../bin/win64 -Dwrapper.key=nHS3sBcSd6ZaDpiu -Dwrapper.port=32000 -Dwrapper
.jvm.port.min=31000 -Dwrapper.jvm.port.max=31999 -Dwrapper.pid=8028 -Dwrapper.version=3.2.3 -Dwrapper.native_library=wrapper -Dwrapper.cpu.timeout=10 -Dwrapper.jvmid=1
jvm 1    | Extensions classpath:
jvm 1    |   [..\..\lib,..\..\lib\camel,..\..\lib\optional,..\..\lib\web,..\..\lib\extra]
jvm 1    | ACTIVEMQ_HOME: ..\..
jvm 1    | ACTIVEMQ_BASE: ..\..
jvm 1    | ACTIVEMQ_CONF: ..\..\conf
jvm 1    | ACTIVEMQ_DATA: ..\..\data
jvm 1    | Loading message broker from: xbean:activemq.xml
jvm 1    |  INFO | Refreshing org.apache.activemq.xbean.XBeanBrokerFactory$1@21f4075c: startup date [Mon Aug 24 14:09:22 CST 2015]; root of context hierarchy
jvm 1    |  INFO | PListStore:[E:\apache-activemq-5.12.0\bin\win64\..\..\data\localhost\tmp_storage] started
jvm 1    |  INFO | Using Persistence Adapter: KahaDBPersistenceAdapter[E:\apache-activemq-5.12.0\bin\win64\..\..\data\kahadb]
jvm 1    |  INFO | Apache ActiveMQ 5.12.0 (localhost, ID:shy-PC-64231-1440396565339-0:1) is starting
jvm 1    |  INFO | Listening for connections at: tcp://shy-PC:61616?maximumConnections=1000&wireFormat.maxFrameSize=104857600
jvm 1    |  INFO | Connector openwire started
jvm 1    |  INFO | Listening for connections at: amqp://shy-PC:5672?maximumConnections=1000&wireFormat.maxFrameSize=104857600
jvm 1    |  INFO | Connector amqp started
jvm 1    |  INFO | Listening for connections at: stomp://shy-PC:61613?maximumConnections=1000&wireFormat.maxFrameSize=104857600
jvm 1    |  INFO | Connector stomp started
jvm 1    |  INFO | Listening for connections at: mqtt://shy-PC:1883?maximumConnections=1000&wireFormat.maxFrameSize=104857600
jvm 1    |  INFO | Connector mqtt started
jvm 1    | {}
jvm 1    |  INFO | Listening for connections at ws://shy-PC:61614?maximumConnections=1000&wireFormat.maxFrameSize=104857600
jvm 1    |  INFO | Connector ws started
jvm 1    |  INFO | Apache ActiveMQ 5.12.0 (localhost, ID:shy-PC-64231-1440396565339-0:1) started
jvm 1    |  INFO | For help or more information please see: http://activemq.apache.org
jvm 1    |  INFO | ActiveMQ WebConsole available at http://0.0.0.0:8161/
jvm 1    |  INFO | ActiveMQ Jolokia REST API available at http://0.0.0.0:8161/api/jolokia/
jvm 1    |  INFO | Initializing Spring FrameworkServlet 'dispatcher'
jvm 1    |  INFO | jolokia-agent: No access restrictor found at classpath:/jolokia-access.xml, access to all MBeans is allowed

        用户配置文件在apache-activemq-5.12.0\conf\users.properties,默认用户密码为admin/admin

        启动ActiveMQ以后,登陆:http://localhost:8161/admin/

        

        打开eclipse,创建两个项目(producer、customer),将apache-activemq-5.5.1\lib下的jar拷贝到项目里并buildpath。分别在两个项目里对应建立如下类:

        producer类(producer项目

package producer;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

public class Producer {
	private static final int SEND_NUMBER = 5;

    public static void main(String[] args) {
        // ConnectionFactory :连接工厂,JMS 用它创建连接
        ConnectionFactory connectionFactory;
        // Connection :JMS 客户端到JMS Provider 的连接
        Connection connection = null;
        // Session: 一个发送或接收消息的线程
        Session session;
        // Destination :消息的目的地;消息发送给谁.
        Destination destination;
        // MessageProducer:消息发送者
        MessageProducer producer;
        // TextMessage message;
        // 构造ConnectionFactory实例对象,此处采用ActiveMq的实现jar
        connectionFactory = new ActiveMQConnectionFactory(
                ActiveMQConnection.DEFAULT_USER,
                ActiveMQConnection.DEFAULT_PASSWORD,
                "tcp://localhost:61616");
        try {
            // 构造从工厂得到连接对象
            connection = connectionFactory.createConnection();
            // 启动
            connection.start();
            // 获取操作连接
            session = connection.createSession(Boolean.TRUE,
                    Session.AUTO_ACKNOWLEDGE);
            // 获取session注意参数值xingbo.xu-queue是一个服务器的queue,须在在ActiveMq的console配置
            destination = session.createQueue("FirstQueue");
            // 得到消息生成者【发送者】
            producer = session.createProducer(destination);
            // 设置不持久化,此处学习,实际根据项目决定
            producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
            // 构造消息,此处写死,项目就是参数,或者方法获取
            sendMessage(session, producer);
            session.commit();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                if (null != connection)
                    connection.close();
            } catch (Throwable ignore) {
            }
        }
    }

    public static void sendMessage(Session session, MessageProducer producer)
            throws Exception {
        for (int i = 1; i <= SEND_NUMBER; i++) {
            TextMessage message = session
                    .createTextMessage("ActiveMq 发送的消息" + i);
            // 发送消息到目的地方
            System.out.println("发送消息:" + "ActiveMq 发送的消息" + i);
            producer.send(message);
        }
    }
}

        customer类(customer项目

package customer;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.MessageConsumer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

public class Customer {
	public static void main(String[] args) {
        // ConnectionFactory :连接工厂,JMS 用它创建连接
        ConnectionFactory connectionFactory;
        // Connection :JMS 客户端到JMS Provider 的连接
        Connection connection = null;
        // Session: 一个发送或接收消息的线程
        Session session;
        // Destination :消息的目的地;消息发送给谁.
        Destination destination;
        // 消费者,消息接收者
        MessageConsumer consumer;
        connectionFactory = new ActiveMQConnectionFactory(
                ActiveMQConnection.DEFAULT_USER,
                ActiveMQConnection.DEFAULT_PASSWORD,
                "tcp://localhost:61616");
        try {
            // 构造从工厂得到连接对象
            connection = connectionFactory.createConnection();
            // 启动
            connection.start();
            // 获取操作连接
            session = connection.createSession(Boolean.FALSE,
                    Session.AUTO_ACKNOWLEDGE);
            // 获取session注意参数值xingbo.xu-queue是一个服务器的queue,须在在ActiveMq的console配置
            destination = session.createQueue("FirstQueue");
            consumer = session.createConsumer(destination);
            while (true) {
                //设置接收者接收消息的时间,为了便于测试,这里谁定为100s
                TextMessage message = (TextMessage) consumer.receive(100000);
                if (null != message) {
                    System.out.println("收到消息" + message.getText());
                } else {
                    break;
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                if (null != connection)
                    connection.close();
            } catch (Throwable ignore) {
            }
        }
    }
}

        然后运行producer项目,以下为控制台输出:

发送消息:ActiveMq 发送的消息1
发送消息:ActiveMq 发送的消息2
发送消息:ActiveMq 发送的消息3
发送消息:ActiveMq 发送的消息4
发送消息:ActiveMq 发送的消息5


收到消息ActiveMq 发送的消息1
收到消息ActiveMq 发送的消息2
收到消息ActiveMq 发送的消息3
收到消息ActiveMq 发送的消息4
收到消息ActiveMq 发送的消息5

       在刚刚打开的页面里找到queues,点击可以看到我们刚刚创建的queues

         注:有时候并不明白队列到底是被用来做什么?或者说它可以做什么?什么情况下需要做?举几个简单的例子,或许不合适,但应该能够明白:

        1、日志。当多个类实例同时产生日志的时候,因为日志文件只有一个,每次写入都只能有一个类实例能够写入,其他全部阻塞。这样显然是消耗性能的,那么如果这个时候有个中间件,类实例只需要把要写的日志传入中间件,就执行结束,然后另外一个程序不断从中间件获取日志再去写入日志,那么这样能够提升很大的性能。

        2、分布式事物。我们知道分布式项目的事物是很麻烦的,加入我们购买了一个商品,那么就需要同时去更改很多东西,比如客户信息、订单信息、发货信息等,在dao或service里一个一个去连接再写入,会导致页面长时间无反应,用户体验度非常差,而如果把这些要添加的数据写到一个中间件,其他专门的程序去读取处理,这样页面就不会出现长时间等待问题。  

        另外注意,TextMessage类的setObjectProperty只能传递原始类型、字符、map、list,其他类型会报错。

与spring集成点击此:http://my.oschina.net/shyloveliyi/blog/496489

© 著作权归作者所有

共有 人打赏支持
尚浩宇
粉丝 53
博文 140
码字总数 112241
作品 3
朝阳
程序员
ActiveMQ 持久化(数据库),查询队列剩余消息数、出队数的实现

《ActiveMQ 持久化(文件),查询队列剩余消息数、出队数的实现》分析了消息队列持久化保存,假如activemq服务器突然停止,服务器启动后,还可以继续查找队列中的消息。现在分析队列中的消息...

cookqq
2013/03/04
0
0
ActiveMQ 持久化(文件),查询队列剩余消息数、出队数的实现

本人博客开始迁移,博客整个架构自己搭建及编码 http://www.cookqq.com/listBlog.action 《ActiveMQ发消息和收消息》详细介绍了ActiveMQ发消息和收消息,消息保存在消息队列(queue)中,消息...

cookqq
2013/03/03
0
1
ActiveMQ 嵌入Tomcat

在一些项目中,单独开启一个ActiveMQ,对于项目实施来说有时略显繁琐。所以我们将ActiveMQ内嵌到Tomcat,Tomcat启动同时就顺带启动了ActiveMQ。由此我们需要掌握三个个重要的知识点 ActiveMQ...

_herbert
08/03
0
0
.Net平台下ActiveMQ入门实例

1.ActiveMQ简介 先分析这么一个场景:当我们在网站上购物时,必须经过,下订单、发票创建、付款处理、订单履行、航运等。但是,当用户下单后,立即跳转到“感谢那您的订单” 页面。不仅如此,...

postdep
2015/08/24
0
0
ActiveMQ集群方案(下)

版权声明:欢迎转载,但是看在我辛勤劳动的份上,请注明来源:http://blog.csdn.net/yinwenjie(未经允许严禁用于商业用途!) https://blog.csdn.net/yinwenjie/article/details/51205822 目...

yunlielai
04/15
0
0

没有更多内容

加载失败,请刷新页面

加载更多

Apache Bench学习笔记

使用apache bench测试并发请求 ab -H "X-IMATRIX-ACCESS-TOKEN:1234567" -c 1000 -n 1000 http://localhost:8080/portfolioes/1/performance...

OSC_fly
29分钟前
3
0
Oracle推出轻量级Java微服务框架Helidon

近日,Oracle 推出 了一个新的开源框架 Helidon ,该项目是一个用于创建基于微服务的应用程序的Java库集合。和 Payara Micro 、 Thorntail (之前的 WildFly Swarm )、 OpenLiberty 、TomEE...

小刀爱编程
30分钟前
5
0
mysql 按周统计

方法一 : 使用 DATE_FORMAT 格式化时间,格式化参数参考 -- 参考地址 SELECTDATE_FORMAT( create_time, '%Y%u' ) weeks,DATE_FORMAT( date_sub( create_time, INTERVAL WEEKDAY( cre...

小鸟00
34分钟前
3
0
深入理解JAVA锁的机制

1. synchronized实现原理 在java代码中使用synchronized可是使用在代码块和方法中,根据Synchronized用的位置可以有这些使用场景: 如图,synchronized可以用在方法上也可以使用在代码块中,...

laigous
36分钟前
2
0
Mysql几种索引类型的区别及适用情况

如大家所知道的,Mysql目前主要有以下几种索引类型:FULLTEXT,HASH,BTREE,RTREE。 那么,这几种索引有什么功能和性能上的不同呢? FULLTEXT 即为全文索引,目前只有MyISAM引擎支持。其可以...

ZhangLG
45分钟前
2
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部