抓住自己最有兴趣的东西,由浅入深,循序渐进地学……
DBMNG数据库管理与应用
当前位置:首页 > 经验分享 > Java组件
大型互联网高并发解决方案之消息中间件技术-activeMQ详解

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: 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配置文件中修改:

案例

在下载的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。因此主要的几个步骤如下:

获得JMS connection factory. 通过我们提供特定环境的连接信息来构造factory。

利用factory构造JMS connection

启动connection

通过connection创建JMS session.

指定JMS destination.

创建JMS producer或者创建JMS message并提供destination.

创建JMS consumer或注册JMS message listener.

发送和接收JMS message.

关闭所有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(HashMaprequestParam) 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();

HashMaprequestParam = new HashMap();

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的请求处理:

import java.util.Map;

import javax.jms.MessageConsumer;

public class RequestProcessor

public void requestHandler(HashMap

System.out.println("requestHandler....."+requestParam.toString());

for(Map.Entryentry : requestParam.entrySet())

System.out.println(entry.getKey()+":"+entry.getValue());

public static void main(String[] args) throws Exception

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);

requestParam = (HashMap) 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.....{朱小厮=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.JMSException;

import javax.jms.TextMessage;

import javax.jms.Topic;

import org.apache.camel.Produce;

public class TopicRequest

//Connection:JMS客户端到JMS Provider的连接

private Connection connection;

//从构造工厂中得到连接对象

connection = connectionFactory.createConnection();

session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);

Topic topic = session.createTopic("MessageTopic");

producer = session.createProducer(topic);

public void submit(String mess) throws Exception

TextMessage message = session.createTextMessage();

message.setText(mess);

public void close()

try

if(session != null)

session.close();

if(producer != null)

producer.close();

if(connection !=null )

connection.close();

catch (JMSException e)

e.printStackTrace();

TopicRequest topicRequest = new TopicRequest();

topicRequest.init();

topicRequest.submit("I'm first");

topicRequest.close();

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

import javax.jms.Message;

import javax.jms.MessageListener;

public class TopicReceive

private MessageConsumer consumer;

consumer = session.createConsumer(topic);

consumer.setMessageListener(new MessageListener(){

@Override

public void onMessage(Message message)

TextMessage tm = (TextMessage) message;

System.out.println(tm);

System.out.println(tm.getText());

});

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



本站文章内容,部分来自于互联网,若侵犯了您的权益,请致邮件chuanghui423#sohu.com(请将#换为@)联系,我们会尽快核实后删除。
关于我们 - 版权声明 - 服务条款 - RSS订阅 - 工作机会 - 友情链接 - 联系我们 - 在线留言
Copyright © 2006-2018 DBMNG.COM All Rights Reserved. Powered by DEVSOARTECH   豫ICP备11002312号-2

豫公网安备 41010502002439号