ActiveMQ简述

概述

ActiveMQ是Apache所提供的一个开源的消息系统，完全采用Java来实现，因此，它能很好地支持J2EE提出的JMS（Java Message Service,即Java消息服务）规范。JMS是一组Java应用程序接口，它提供消息的创建、发送、读取等一系列服务。JMS提供了一组公共应用程序接口和响应的语法，类似于Java数据库的统一访问接口JDBC,它是一种与厂商无关的API，使得Java程序能够与不同厂商的消息组件很好地进行通信。

JMS支持两种消息发送和接收模型。一种称为P2P(Ponit to Point)模型，即采用点对点的方式发送消息。P2P模型是基于队列的，消息生产者发送消息到队列，消息消费者从队列中接收消息，队列的存在使得消息的异步传输称为可能，P2P模型在点对点的情况下进行消息传递时采用。

另一种称为Pub/Sub(Publish/Subscribe，即发布-订阅)模型，发布-订阅模型定义了如何向一个内容节点发布和订阅消息，这个内容节点称为topic(主题)。主题可以认为是消息传递的中介，消息发布这将消息发布到某个主题，而消息订阅者则从主题订阅消息。主题使得消息的订阅者与消息的发布者互相保持独立，不需要进行接触即可保证消息的传递，发布-订阅模型在消息的一对多广播时采用。

ActiveMQ的安装

下载最新的安装包apache-activemq-5.13.2-bin.tar.gz（此包linux下的，案例也是针对linux系统进行阐述，当然ActiveMQ也有win版的，这里就不赘述了），可以去官网下载，也可以在下方留言区留下你的邮箱，博主会发给你的~

下载之后解压： tar -zvxf apache-activemq-5.13.2-bin.tar.gz

ActiveMQ目录内容有：

  • bin目录包含ActiveMQ的启动脚本
  • conf目录包含ActiveMQ的所有配置文件
  • data目录包含日志文件和持久性消息数据
  • example: ActiveMQ的示例
  • lib: ActiveMQ运行所需要的lib
  • webapps: ActiveMQ的web控制台和一些相关的demo

运行命令：activemq start（在activemq/bin下运行）

INFO: Loading ‘/users/shr/apache-activemq-5.13.2//bin/env‘
INFO: Using java ‘/users/shr/util/JavaDir/jdk/bin/java‘
INFO: Starting - inspect logfiles specified in logging.properties and log4j.properties to get details
INFO: pidfile created : ‘/users/shr/apache-activemq-5.13.2//data/activemq.pid‘ (pid ‘986‘)

查看activemq是否运行命令：ps -aux | grep activemq

shr        986  1.2  9.7 1281720 201936 pts/5  Sl   19:43   0:17 /users/shr/util/JavaDir/jdk/bin/java -Xms64M -Xmx1G -Djava.util.logging.config.file=logging.properties -Djava.security.auth.login.config=/users/shr/apache-activemq-5.13.2//conf/login.config -Dcom.sun.management.jmxremote -Djava.awt.headless=true -Djava.io.tmpdir=/users/shr/apache-activemq-5.13.2//tmp -Dactivemq.classpath=/users/shr/apache-activemq-5.13.2//conf:/users/shr/apache-activemq-5.13.2//../lib/: -Dactivemq.home=/users/shr/apache-activemq-5.13.2/ -Dactivemq.base=/users/shr/apache-activemq-5.13.2/ -Dactivemq.conf=/users/shr/apache-activemq-5.13.2//conf -Dactivemq.data=/users/shr/apache-activemq-5.13.2//data -jar /users/shr/apache-activemq-5.13.2//bin/activemq.jar start
shr       1501  0.0  0.0   5176   724 pts/5    S+   20:06   0:00 grep activemq

关闭命令： activemq stop

INFO: Loading ‘/users/shr/apache-activemq-5.13.2//bin/env‘
INFO: Using java ‘/users/shr/util/JavaDir/jdk/bin/java‘
INFO: Waiting at least 30 seconds for regular process termination of pid ‘986‘ :
Java Runtime: Oracle Corporation 1.7.0_79 /users/shr/util/JavaDir/jdk1.7.0_79/jre
  Heap sizes: current=63232k  free=62218k  max=932096k
    JVM args: -Xms64M -Xmx1G -Djava.util.logging.config.file=logging.properties -Djava.security.auth.login.config=/users/shr/apache-activemq-5.13.2//conf/login.config -Dactivemq.classpath=/users/shr/apache-activemq-5.13.2//conf:/users/shr/apache-activemq-5.13.2//../lib/: -Dactivemq.home=/users/shr/apache-activemq-5.13.2/ -Dactivemq.base=/users/shr/apache-activemq-5.13.2/ -Dactivemq.conf=/users/shr/apache-activemq-5.13.2//conf -Dactivemq.data=/users/shr/apache-activemq-5.13.2//data
Extensions classpath:
  [/users/shr/apache-activemq-5.13.2/lib,/users/shr/apache-activemq-5.13.2/lib/camel,/users/shr/apache-activemq-5.13.2/lib/optional,/users/shr/apache-activemq-5.13.2/lib/web,/users/shr/apache-activemq-5.13.2/lib/extra]
ACTIVEMQ_HOME: /users/shr/apache-activemq-5.13.2
ACTIVEMQ_BASE: /users/shr/apache-activemq-5.13.2
ACTIVEMQ_CONF: /users/shr/apache-activemq-5.13.2/conf
ACTIVEMQ_DATA: /users/shr/apache-activemq-5.13.2/data
Connecting to pid: 986
..Stopping broker: localhost
.. TERMINATED

ActiveMQ的默认服务端口为61616，这个可以在conf/activemq.xml配置文件中修改：

<transportConnectors>
    <!-- DOS protection, limit concurrent connections to 1000 and frame size to 100MB -->
    <transportConnector name="openwire" uri="tcp://0.0.0.0:61616?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
</transportConnectors>

案例

在下载的apache-activemq-5.13.2-bin.tar.gz包中解压有一个jar包：activemq-all-5.13.2.jar，引入这个jar到你的项目中即可开始编写案例代码。

博主的activemq服务器地址为10.10.195.187，这个在下面代码中会有体现。

按照JMS的规范，我们首先需要获得一个JMS connection factory.，通过这个connection factory来创建connection.在这个基础之上我们再创建session, destination, producer和consumer。因此主要的几个步骤如下：

  1. 获得JMS connection factory. 通过我们提供特定环境的连接信息来构造factory。
  2. 利用factory构造JMS connection
  3. 启动connection
  4. 通过connection创建JMS session.
  5. 指定JMS destination.
  6. 创建JMS producer或者创建JMS message并提供destination.
  7. 创建JMS consumer或注册JMS message listener.
  8. 发送和接收JMS message.
  9. 关闭所有JMS资源，包括connection, session, producer, consumer等。

下面来看代码举例（P2P式）。

通过Java实现的基于ActiveMQ的请求提交：

package com.zzh.activemq;

import java.io.Serializable;
import java.util.HashMap;

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

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

public class RequestSubmit
{
    //消息发送者
    private MessageProducer producer;
    //一个发送或者接受消息的线程
    private Session session;

    public void init() throws Exception
    {
        //ConnectionFactory连接工厂，JMS用它创建连接
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
                ActiveMQConnection.DEFAULT_USER,
                ActiveMQConnection.DEFAULT_PASSWORD,
                "tcp://10.10.195.187:61616");
        //Connection：JMS客户端到JMS Provider的连接，从构造工厂中得到连接对象
        Connection connection = connectionFactory.createConnection();
        //启动
        connection.start();
        //获取连接操作
        session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
        Destination destinatin = session.createQueue("RequestQueue");
        //得到消息生成（发送）者
        producer = session.createProducer(destinatin);
        //设置不持久化
        producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
    }

    public void submit(HashMap<Serializable,Serializable> requestParam) throws Exception
    {
        ObjectMessage message = session.createObjectMessage(requestParam);
        producer.send(message);
        session.commit();
    }

    public static void main(String[] args) throws Exception{
        RequestSubmit submit = new RequestSubmit();
        submit.init();
        HashMap<Serializable,Serializable> requestParam = new HashMap<Serializable,Serializable>();
        requestParam.put("朱小厮", "zzh");
        submit.submit(requestParam);
    }
}

创建Session时有两个非常重要的参数，第一个boolean类型的参数用来表示是否采用事务消息。如果是事务消息，对于的参数设置为true，此时消息的提交自动有comit处理，消息的回滚则自动由rollback处理。加入消息不是事务的，则对应的该参数设置为false，此时分为三种情况：

  • Session.AUTO_ACKNOWLEDGE表示Session会自动确认所接收到的消息。
  • Session.CLIENT_ACKNOWLEDGE表示由客户端程序通过调用消息的确认方法来确认所接收到的消息。
  • Session.DUPS_OK_ACKNOWLEDGE使得Session将“懒惰”地确认消息，即不会立即确认消息，这样有可能导致消息重复投递。

提供Java实现的基于ActiveMQ的请求处理：

package com.zzh.activemq;

import java.io.Serializable;
import java.util.HashMap;
import java.util.Map;

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

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

public class RequestProcessor
{
    public void requestHandler(HashMap<Serializable,Serializable> requestParam) throws Exception
    {
        System.out.println("requestHandler....."+requestParam.toString());
        for(Map.Entry<Serializable, Serializable> entry : requestParam.entrySet())
        {
            System.out.println(entry.getKey()+":"+entry.getValue());
        }
    }

    public static void main(String[] args) throws Exception
    {
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
                ActiveMQConnection.DEFAULT_USER,
                ActiveMQConnection.DEFAULT_PASSWORD,
                "tcp://10.10.195.187:61616");
        Connection connection = connectionFactory.createConnection();
        connection.start();
        Session session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
        Destination destination = session.createQueue("RequestQueue");
        //消息消费（接收）者
        MessageConsumer consumer = session.createConsumer(destination);

        RequestProcessor processor = new RequestProcessor();

        while(true)
        {
            ObjectMessage message = (ObjectMessage) consumer.receive(1000);
            if(null != message)
            {
                System.out.println(message);
                HashMap<Serializable,Serializable> requestParam = (HashMap<Serializable,Serializable>) message.getObject();
                processor.requestHandler(requestParam);
            }
            else
            {
                break;
            }
        }
    }
}

输出结果：

ActiveMQObjectMessage {commandId = 6, responseRequired = false, messageId = ID:hidden-PC-58748-1460550507055-1:1:1:1:1, originalDestination = null, originalTransactionId = null, producerId = ID:hidden-PC-58748-1460550507055-1:1:1:1, destination = queue://RequestQueue, transactionId = TX:ID:hidden-PC-58748-1460550507055-1:1:1, expiration = 0, timestamp = 1460550507333, arrival = 0, brokerInTime = 1460550505969, brokerOutTime = 1460550509143, correlationId = null, replyTo = null, persistent = false, type = null, priority = 4, groupID = null, groupSequence = 0, targetConsumerId = null, compressed = false, userID = null, content = [email protected]74a456bb, marshalledProperties = null, dataStructure = null, redeliveryCounter = 0, size = 0, properties = null, readOnlyProperties = true, readOnlyBody = true, droppable = false, jmsXGroupFirstForConsumer = false}
requestHandler.....{朱小厮=zzh}
朱小厮:zzh

可以通过页面查看队列的使用情况，在浏览器中输入http://10.10.195.187:8161/admin/queues.jsp，用户名和密码都是：admin，看到以下页面：

这个是在jetty服务器下跑的，可以修改conf/jetty.xml来修改相关jetty配置。

上面的例子是关于P2P模式的，不过有个不妥之处，就是没有资源的释放。下面举一个Pub/Sub模式的。

通过JMS创建ActiveMQ的topic，并给topic发送消息：

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.ObjectMessage;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;

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

public class TopicRequest
{
    //消息发送者
    private MessageProducer producer;
    //一个发送或者接受消息的线程
    private Session session;
    //Connection：JMS客户端到JMS Provider的连接
    private Connection connection;

    public void init() throws Exception
    {
        //ConnectionFactory连接工厂，JMS用它创建连接
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
                ActiveMQConnection.DEFAULT_USER,
                ActiveMQConnection.DEFAULT_PASSWORD,
                "tcp://10.10.195.187:61616");
        //从构造工厂中得到连接对象
        connection = connectionFactory.createConnection();
        //启动
        connection.start();
        //获取连接操作
        session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
        Topic topic = session.createTopic("MessageTopic");
        producer = session.createProducer(topic);
        //设置不持久化
        producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
    }

    public void submit(String mess) throws Exception
    {
        TextMessage message = session.createTextMessage();
        message.setText(mess);
        producer.send(message);
    }

    public void close()
    {
        try
        {
            if(session != null)
                session.close();
            if(producer != null)
                producer.close();
            if(connection !=null )
                connection.close();
        }
        catch (JMSException e)
        {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) throws Exception
    {
        TopicRequest topicRequest = new TopicRequest();
        topicRequest.init();
        topicRequest.submit("I‘m first");
        topicRequest.close();
    }
}

消息发送到对应的topic后，需要将listener注册到需要订阅的topic上，以便能够接收该topic的消息：

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;

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

public class TopicReceive
{
    private MessageConsumer consumer;
    private Session session;

    public void init() throws Exception
    {
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
                ActiveMQConnection.DEFAULT_USER,
                ActiveMQConnection.DEFAULT_PASSWORD,
                "tcp://10.10.195.187:61616");
        Connection connection = connectionFactory.createConnection();
        connection.start();
        session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
        Topic topic = session.createTopic("MessageTopic");
        consumer = session.createConsumer(topic);

        consumer.setMessageListener(new MessageListener(){
            @Override
            public void onMessage(Message message)
            {
                TextMessage tm = (TextMessage) message;
                System.out.println(tm);
                try
                {
                    System.out.println(tm.getText());
                }
                catch (JMSException e)
                {
                    e.printStackTrace();
                }
            }
        });
    }

    public static void main(String[] args) throws Exception
    {
        TopicReceive receive = new TopicReceive();
        receive.init();
    }
}

输出结果：

ActiveMQTextMessage {commandId = 5, responseRequired = false, messageId = ID:hidden-PC-50073-1460597487065-1:1:1:1:1, originalDestination = null, originalTransactionId = null, producerId = ID:hidden-PC-50073-1460597487065-1:1:1:1, destination = topic://MessageTopic, transactionId = null, expiration = 0, timestamp = 1460597487308, arrival = 0, brokerInTime = 1460597487297, brokerOutTime = 1460597487298, correlationId = null, replyTo = null, persistent = false, type = null, priority = 4, groupID = null, groupSequence = 0, targetConsumerId = null, compressed = false, userID = null, content = [email protected]2e4d3abf, marshalledProperties = null, dataStructure = null, redeliveryCounter = 0, size = 0, properties = null, readOnlyProperties = true, readOnlyBody = true, droppable = false, jmsXGroupFirstForConsumer = false, text = I‘m first}
I‘m first

参考文献

1. 《大型分布式网站架构——设计与实践》陈康贤著。

2. http://activemq.apache.org/

时间： 2024-10-12 17:12:06

ActiveMQ简述的相关文章

Sping+ActiveMQ整合

通过前一篇<ActiveMQ简述>大概对ActiveMQ有了一个大概的认识,本篇所阐述的是如何通过Spring继承ActiveMQ进而更有效.更灵活的运用ActiveMQ. Spring和ActiveMQ整合需要在项目中包含以下这几个jar包(缺一不可):activeio-core-3.1.4.jar,activemq-all-5.13.2.jar,activemq-pool-5.13.2.jar,commons-pool2-2.4.2.jar,这些jar可以在ActiveMQ的安装包中的/l

消息队列之ActiveMQ安装配置

简述:ActiveMQ是由Apache出品的,一款最流行的,能力强劲的开源消息总线.ActiveMQ是一个完全支持JMS1.1和J2EE 1.4规范的 JMS Provider实现,它非常快速,支持多种语言的客户端和协议,而且可以非常容易的嵌入到企业的应用环境中,并有许多高级功能. 一.安装ActiveMQ 1.下载ActiveMQ 官方网站:http://activemq.apache.org/activemq-5122-release.html 2.安装JDK7,参考(http://ljoh

框架面试题（maven、ZooKeeper、Dubbo、Nginx、Redis、Lucene、Solr、ActiveMQ、JMS

什么是 Maven? Maven 使用项目对象模型(POM)的概念,可以通过一小段描述信息来管理项目的构建, 报告和文档的软件项目管理工具. Maven 除了以程序构建能力为特色之外,还提供高级项目管理工具.由于 Maven 的 缺省构建规则有较高的可重用性,所以常常用两三行 Maven 构建脚本就可以构建简单的项 目.由于 Maven 的面向项目的方法,许多 Apache Jakarta 项目发布时使用 Maven,而 且公司项目采用 Maven 的比例在持续增长. Maven 的出现,解决了

Hibernate简述及入门实例

一.Hibernate简述 总的概括,Hibernate是一个ORM的轻量级持久层框架,解决了对象和关系数据库中表的不匹配问题(阻抗不匹配)以及拥有开发代码不用去继承hibernate类或接口的优势(无侵入性).hibernate框架实现使得开发人员可以避免反复地编写javajdbc部分代码,应用面向对象的思维操作关系型数据库. 二.使用myeclipse创建hibernate实例两种方法(以hibernate3.5.2及mysql为例) a)手动编写hibernate.cfg.xml及*.hb

ActiveMQ集群Master-Slave + Broker Cluster模式

一.简介 Master-Slave集群: 由至少3个节点组成,一个Master节点,其他为Slave节点.只有Master节点对外提供服务,Slave节点处于等待状态.当主节点宕机后,从节点会推举出一个节点出来成为新的Master节点,继续提供服务. 优点是可以解决多服务热备的高可用问题,缺点是无法解决负载均衡和分布式的问题. Broker Cluster集群: Broker-Cluster部署方式中,各个broker通过网络互相连接,并共享queue.当broker-A上面指定的queue-A

rabbitMQ、activeMQ、zeroMQ、Kafka、Redis 比较

Kafka作为时下最流行的开源消息系统,被广泛地应用在数据缓冲.异步通信.汇集日志.系统解耦等方面.相比较于RocketMQ等其他常见消息系统,Kafka在保障了大部分功能特性的同时,还提供了超一流的读写性能. 针对Kafka性能方面进行简单分析,相关数据请参考:https://segmentfault.com/a/1190000003985468,下面介绍一下Kafka的架构和涉及到的名词: Topic:用于划分Message的逻辑概念,一个Topic可以分布在多个Broker上. Parti

在Spring下集成ActiveMQ

1.参考文献 Spring集成ActiveMQ配置 Spring JMS异步发收消息 ActiveMQ 2.环境 在前面的一篇ActiveMQ入门实例中我们实现了消息的异步传送,这篇博文将如何在spring环境下集成ActiveMQ.如果要在spring下集成ActiveMQ,那么就需要将如下jar包导入项目: 本文有两篇参考文献,因此有两个实例,项目结构如下图所示: 3.实例1 信息发送者:HelloSender.java package edu.sjtu.erplab.springactiv

Spring Boot集成ActiveMQ

在Spring Boot中集成ActiveMQ相对还是比较简单的,都不需要安装什么服务,默认使用内存的activeMQ,当然配合ActiveMQ Server会更好.在这里我们简单介绍怎么使用,本节主要分以下几个步骤: (1) 新建Maven Java Project; (2) 在pom.xml引入依赖: (3) 编码测试 (4) 配置信息 接下来看看各个步骤的操作: (1) 新建Maven Java Project; 新建一个工程取名为spring-boot-activemq (2) 在pom

Struts2基本使用（二）--配置文件简述

配置文件简述 引入Struts2框架之后项目中多了一个struts.xml,以及web.xml也多了一些代码 web.xml只要的功能就是拦截用户的请求其多出的代码如下: <filter> <filter-name>struts2</filter-name> <filter-class>org.apache.struts2.dispatcher.ng.filter.StrutsPrepareAndExecuteFilter</filter-class&