文档章节

ActiveMQ消息系统研究与学习

Yemon
 Yemon
发布于 2016/05/17 10:32
字数 2312
阅读 14
收藏 0
点赞 2
评论 0

概述

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&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("joker", "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 = org.apache.activemq.util.ByteSequence@74a456bb, marshalledProperties = null, dataStructure = null, redeliveryCounter = 0, size = 0, properties = null, readOnlyProperties = true, readOnlyBody = true, droppable = false, jmsXGroupFirstForConsumer = false}
requestHandler.....

可以通过页面查看队列的使用情况,在浏览器中输入页面  ,用户名和密码都是: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 = org.apache.activemq.util.ByteSequence@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



本文转载自:http://blog.csdn.net/hj7jay/article/details/51422543

共有 人打赏支持
Yemon
粉丝 10
博文 320
码字总数 22799
作品 0
广州
架构师
ActiveMQ集群方案(下)

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

yunlielai ⋅ 04/15 ⋅ 0

ActiveMQ集群方案(上)

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

yunlielai ⋅ 04/15 ⋅ 0

ActiveMQ初探(2)——与Spring整合

在上一章:ActiveMQ(1)——介绍与基本使用中,我们通过测试方法熟悉了ActiveMQ的流程,本章将ActiveMQ与Spring进行整合。 与Spring整合,除了原本的外,还需导入和包,如果Spring为5.0+,需...

yuanlaijike ⋅ 05/08 ⋅ 0

ActiveMQ初探(1)——介绍与基本使用

一、ActiveMQ 1.1 什么是ActiveMQ 是Apache出品,最流行的,能力强劲的。ActiveMQ是一个完全支持和规范的 实现,尽管规范出台已经是很久的事情了,但是JMS在当今的J2EE应用中间仍然扮演着特殊...

yuanlaijike ⋅ 04/15 ⋅ 0

消息队列篇—详谈ActiveMQ消息队列模式的分析及使用

消息队列(Message Queue)是分布式系统中重要的组件,通用使用场景可以简单地描述为当不需要立即获得结果,但是并发量需控制时就需要使用消息队列。消息列队有两种消息模式,一种是点对点的...

afreon ⋅ 04/22 ⋅ 0

爬虫架构 | 消息队列应用场景及ActiveMQ、RabbitMQ、RocketMQ、Kafka对比

前言:在之前的业务中,使用了Kafka和RabbitMQ两种消息队列,这篇文章来做一个总结。 消息队列中间件是分布式系统中重要的组件,主要实现异步消息,应用解耦,流量削峰及消息通讯等功能。 下...

小怪聊职场 ⋅ 04/26 ⋅ 0

activeMQ发送邮件小例子

先安装apache-activemq-5.11.1 ,然后启动activemq.bat 启动activeMQ的控制台:http://localhost:8161/admin/,展示图如下: 解释:Number Of Pending Messages :等待消费的消息,这个是当前...

zhouyuzhoujing ⋅ 05/07 ⋅ 0

ActiveMQ初探(3)——与SpringBoot整合

在上一章:ActiveMQ(2)——与Spring整合中,我们已经完成了ActiveMQ与Spring进行整合,本章将进行与SpringBoot整合。 发送即时消息 Step1:添加依赖包: 这里我们不使用网上大多数的,因为...

yuanlaijike ⋅ 05/08 ⋅ 0

ActiveMq笔记2-消息持久化

为了避免意外宕机以后丢失信息,需要做到重启后可以恢复消息队列,消息系统一般都会采用持久化机制。 ActiveMQ的消息持久化机制有JDBC,AMQ,KahaDB和LevelDB, 无论使用哪种持久化方式,消息...

狂小白 ⋅ 02/25 ⋅ 0

漏网de鱼/fish java分布式商城B2C

系统介绍 Fish 是J2EE分布式开发,技术栈:(dubbo、zookeeper、activemq、Spring、SpringMVC、MyBatis、Shiro、redis、quartz、activiti、MYCAT) 包括核心模块如:用户管理、系统设置、权限...

漏网de鱼 ⋅ 04/23 ⋅ 0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

Cube、Cuboid 和 Cube Segment

1.Cube (或Data Cube),即数据立方体,是一种常用于数据分析与索引的技术;它可以对原始数据建立多维度索引。通过 Cube 对数据进行分析,可以大大加快数据的查询效率 2.Cuboid 在 Kylin 中特...

无精疯 ⋅ 37分钟前 ⋅ 0

github太慢

1:用浏览器访问 IPAddress.com or http://tool.chinaz.com 使用 IP Lookup 工具获得github.com和github.global.ssl.fastly.net域名的ip地址 2:/etc/hosts文件中添加如下格式(IP最好自己查一...

whoisliang ⋅ 39分钟前 ⋅ 0

非阻塞同步之 CAS

为解决线程安全问题,互斥同步相当于以时间换空间。多线程情况下,只有一个线程可以访问同步代码。这种同步也叫阻塞同步(Blocking Synchronization). 这种同步属于一种悲观并发策略。认为只...

长安一梦 ⋅ 50分钟前 ⋅ 0

云计算的选择悖论如何对待?

人们都希望在工作和生活中有所选择。但心理学家的调查研究表明,在多种选项中进行选择并不一定会使人们更快乐,甚至不会产生更好的决策。心理学家Barry Schwartz称之为“选择悖论”。云计算为...

linux-tao ⋅ 52分钟前 ⋅ 0

我的第一篇个人博客

虽然这是个技术博客,但是,我总是想写一些自己的东西,所有就大胆的在这里写下了第一篇非技术博客。技术博客也很久没有更新,个人原因。 以后自己打算在这里写一些非技术博客,可能个人观点...

Mrs_CoCo ⋅ 53分钟前 ⋅ 0

Redis 注册为 Windows 服务

Redis 注册为 Windows 服务 redis 注册为 windows 服务相关命令 注册服务 redis-server.exe –service-install redis.windows.conf 删除服务 redis-server –service-uninstall 启动服务 re......

Os_yxguang ⋅ 53分钟前 ⋅ 0

世界那么大,语言那么多,为什么选择Micropython,它的优势在哪?

最近国内MicroPython风靡程序界,是什么原因导致它这么火呢?是因为他功能强大,遵循Mit协议开源么? 错!因为使用它真的是太舒服了!!! Micropython的由来,这得益于Damien George这位伟大...

bodasisiter ⋅ 57分钟前 ⋅ 0

docker 清理总结

杀死所有正在运行的容器 docker kill $(docker ps -a -q) 删除所有已经停止的容器(docker rm没有加-f参数,运行中的容器不会删掉) docker rm $(docker ps -a -q) 删除所有未打 dangling 标...

vvx1024 ⋅ 今天 ⋅ 0

关于学习

以前学车的时候,教练说了这样的一句话:如果一个人坐在车上一直学,一直学,反而不如大家轮流着学。因为一个人一直学,就没有给自己留空间来反思和改进。而轮流着学的时候大家下来之后思考上...

mskk ⋅ 今天 ⋅ 0

压缩工具之gzip-bzip2-xz

win下常见压缩工具:rar zip 7z linux下常见压缩工具:zip gz bz2 xz tar.gz tar.bz2 tar.xz gzip 不支持目录压缩 gzip 1.txt #压缩。执行后1.txt消失,生成1.txt.gz压缩文件 gzip -d 1.txt....

ZHENG-JY ⋅ 今天 ⋅ 0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

返回顶部
顶部