文档章节

JAVA的activeMQ

zimingforever
 zimingforever
发布于 2012/05/29 12:02
字数 1709
阅读 1134
收藏 3

其实我对activeMQ 一点都不了解,也没用过,而且最近的项目里也不会使用。因为之前听同学说他们正在用,就搞出来看看是什么东西。顺便玩玩。

ActiveMQ 是Apache出品,最流行的,能力强劲的开源消息总线。ActiveMQ 是一个完全支持JMS1.1和J2EE 1.4规范的 JMS Provider实现,尽管JMS规范出台已经是很久的事情了,但是JMS在当今的J2EE应用中间仍然扮演着特殊的地位。

首先是下载部署,常用的目录结构如下

 +bin (windows下面的bat和unix/linux下面的sh)
 +conf (activeMQ配置目录,包含最基本的activeMQ配置文件)
 +data (默认是空的)
 +docs (index,replease版本里面没有文档,-.-b不知道为啥不带)
 +example (几个例子
 +lib (activemMQ使用到的lib)
   -apache-activemq-4.1-incubator.jar (ActiveMQ的binary)
  -LICENSE.txt
  -NOTICE.txt
  -README.txt
  -user-guide.html

然后是部署项目,example里面有现成的程序,放到本地跑跑。

补充一个红薯的文档。

http://www.oschina.net/question/12_19665

关于JMS的东西我在补充些资料,下面是JMS的主要接口

ConnectionFactory 接口(连接工厂) 用户用来创建到JMS提供者的连接的被管对象。JMS客户通过可移植的接口访问连接,这样当下层的实现改变时,代码不需要进行修改。 管理员在JNDI名字空间中配置连接工厂,这样,JMS客户才能够查找到它们。根据消息类型的不同,用户将使用队列连接工厂,或者主题连接工厂。
Connection 接口(连接) 连接代表了应用程序和消息服务器之间的通信链路。在获得了连接工厂后,就可以创建一个与JMS提供者的连接。根据不同的连接类型,连接允许用户创建会话,以发送和接收队列和主题到目标。
Destination 接口(目标) 目标是一个包装了消息目标标识符的被管对象,消息目标是指消息发布和接收的地点,或者是队列,或者是主题。JMS管理员创建这些对象,然后用户通过JNDI发现它们。和连接工厂一样,管理员可以创建两种类型的目标,点对点模型的队列,以及发布者/订阅者模型的主题。
MessageConsumer 接口(消息消费者) 由会话创建的对象,用于接收发送到目标的消息。消费者可以同步地(阻塞模式),或异步(非阻塞)接收队列和主题类型的消息。
MessageProducer 接口(消息生产者) 由会话创建的对象,用于发送消息到目标。用户可以创建某个目标的发送者,也可以创建一个通用的发送者,在发送消息时指定目标。
Message 接口(消息) 是在消费者和生产者之间传送的对象,也就是说从一个应用程序传送到另一个应用程序。一个消息有三个主要部分: 消息头(必须):包含用于识别和为消息寻找路由的操作设置。 一组消息属性(可选):包含额外的属性,支持其他提供者和用户的兼容。可以创建定制的字段和过滤器(消息选择器)。 一个消息体(可选):允许用户创建五种类型的消息(文本消息,映射消息,字节消息,流消息和对象消息)。 消息接口非常灵活,并提供了许多方式来定制消息的内容。
Session 接口(会话) 表示一个单线程的上下文,用于发送和接收消息。由于会话是单线程的,所以消息是连续的,就是说消息是按照发送的顺序一个一个接收的。会话的好处是它支持事务。如果用户选择了事务支持,会话上下文将保存一组消息,直到事务被提交才发送这些消息。在提交事务之前,用户可以使用回滚操作取消这些消息。一个会话允许用户创建消息生产者来发送消息,创建消息消费者来接收消息。
下面是一个JMS的流程图

 

下面是一个测试的代码

public class MyActiveMqTest {

	public static void main(String[] args) {
		Receiver receiver=new Receiver();
		Sender sender =new Sender();
		try {
			sender.send();
			receiver.receive();
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	static class Receiver {
		public static void receive() throws Exception {
			ConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
			Connection connection = connectionFactory.createConnection();
			connection.start();
			final Session session = connection.createSession(Boolean.TRUE,
					Session.AUTO_ACKNOWLEDGE);
			Destination destination = session.createQueue("my-queue");
			MessageConsumer consumer = session.createConsumer(destination);
			/*
			 * //listener 方式 consumer.setMessageListener(new MessageListener() {
			 * 
			 * public void onMessage(Message msg) { MapMessage message =
			 * (MapMessage) msg; //TODO something.... System.out.println("收到消息:"
			 * + new Date(message.getLong("count"))); session.commit(); }
			 * 
			 * }); Thread.sleep(30000);
			 */
			int i = 0;
			while (i < 3) {
				i++;
				MapMessage message = (MapMessage) consumer.receive();
				session.commit();
				// TODO something....
				System.out
						.println("收到消息:" + new Date(message.getLong("count")));
			}

			session.close();
			connection.close();
		}
	}

	static class Sender {
		public static void send() throws Exception {
			ConnectionFactory connectionFactory = new ActiveMQConnectionFactory();

			Connection connection = connectionFactory.createConnection();
			connection.start();

			Session session = connection.createSession(Boolean.TRUE,
					Session.AUTO_ACKNOWLEDGE);
			Destination destination = session.createQueue("my-queue");

			MessageProducer producer = session.createProducer(destination);
			for (int i = 0; i < 3; i++) {
				MapMessage message = session.createMapMessage();
				message.setLong("count", new Date().getTime());
				Thread.sleep(1000);
				// 通过消息生产者发出消息
				System.out.println("发送消息"+i+new Date());
				producer.send(message);
			}
			session.commit();
			session.close();
			connection.close();
		}
	}
}

下面是打印的结果
发送消息0Fri Jun 08 19:41:45 CST 2012
发送消息1Fri Jun 08 19:41:52 CST 2012
发送消息2Fri Jun 08 19:41:55 CST 2012
收到消息:Fri Jun 08 19:41:40 CST 2012
收到消息:Fri Jun 08 19:41:50 CST 2012
收到消息:Fri Jun 08 19:41:53 CST 2012


根据红薯的列子,我还应该试下和spring的结合,搭项目的spring环境。

先贴代码:

public class MyActiveMqSpringTest {

	public static void main(String[] args) {
		Receiver receiver = new Receiver();
		Sender sender = new Sender();
		try {
			sender.send();
			receiver.receive();
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	static class Receiver {
		public static void receive() throws Exception {
			ApplicationContext ctx = new FileSystemXmlApplicationContext(
					"classpath:app*.xml");
			JmsTemplate jmsTemplate = (JmsTemplate) ctx.getBean("jmsTemplate");
			jmsTemplate.send(new MessageCreator() {
				public Message createMessage(Session session)
						throws JMSException {
					MapMessage mm = session.createMapMessage();
					mm.setLong("count", new Date().getTime());
					return mm;
				}

			});
		}
	}

	static class Sender {
		public static void send() throws Exception {
			ApplicationContext ctx = new FileSystemXmlApplicationContext(
					"classpath:app*.xml");
			JmsTemplate jmsTemplate = (JmsTemplate) ctx.getBean("jmsTemplate");
			while (true) {
				Map<String, Object> mm = (Map<String, Object>) jmsTemplate
						.receiveAndConvert();
				System.out.println("收到消息:" + new Date((Long) mm.get("count")));
			}
		}
	}
}

下面是spring的配置

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd">

	<!-- 在非 web / ejb 容器中使用 pool 时,要手动 stop,spring 不会为你执行 destroy-method 的方法 -->
	<!-- 
	<bean id="jmsFactory" class="org.apache.activemq.pool.PooledConnectionFactory"
		destroy-method="stop">
		<property name="connectionFactory">
			<bean class="org.apache.activemq.ActiveMQConnectionFactory">
				<property name="brokerURL" value="tcp://localhost:61616" />
			</bean>
		</property>
	</bean>
	 -->
	<bean id="jmsFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
		<property name="brokerURL" value="tcp://localhost:61616" />
	</bean>
	<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
		<property name="connectionFactory" ref="jmsFactory" />
		<property name="defaultDestination" ref="destination" />
		<property name="messageConverter">
			<bean
				class="org.springframework.jms.support.converter.SimpleMessageConverter" />
		</property>
	</bean>
	<bean id="destination" class="org.apache.activemq.command.ActiveMQQueue">
		<constructor-arg index="0" value="my-queue" />
	</bean>

</beans>

失败了。。没有结果输出。有时间再搞吧。

总结一下,今天主要就是熟悉了下JMS的模型,还知道有个activeMQ的东西,简单的搭建了一下环境,能跑出个helloWorld。ActiveMq和Spring结合那块没有成功,有空再搞搞。

 

 

© 著作权归作者所有

上一篇: tomcat配置数据源
下一篇: 编程字体
zimingforever
粉丝 142
博文 266
码字总数 315040
作品 0
杭州
程序员
私信 提问
加载中

评论(2)

zimingforever
zimingforever 博主

引用来自“心扬”的评论

你妹呀,写了那么长,最后来个失败了

额。。后来没时间搞了。。。。
心扬
你妹呀,写了那么长,最后来个失败了
ActiveMQ安装配置和使用简例

本文作者:Zhang Phil 原文链接:http://blog.csdn.net/zhangphil/article/details/48173665 ActiveMQ安装配置和使用简例 ActiveMQ是一套JMS(Java Message Service)开源消息服务实现的组件...

开开心心过
2015/09/02
0
0
ActiveMQ学习记录 之 消息持久化

1:前言 这一段给公司开发消息总线有机会研究ActiveMQ,今天撰文给大家介绍一下他的持久化消息。本文只介绍三种方式,分别是持久化为文件,MYSql,Oracle。下面逐一介绍。 A:持久化为文件 这...

李格尔楞
2017/11/01
54
0
ubuntu上安装activemq,然后启动不起来,配置文件要改什么地方吗?

我在ubuntu上sudo apt-get intstall activemq 后, 然后到bin下启动activemq:./activemq start INFO: Loading '/usr/share/activemq/activemq-options' INFO: Using java '/usr/bin/java' ......

海空天阔007
2015/12/23
1K
1
JMS配置说明-----activeMQ-5.6

1 简介 activeMQ是一个完全支持JMS1.1 和J2EE规范的JMS Provider实现; 尽管规范出台已经是很久的事情了,但JMS在当今的J2EE应用中仍然扮演着特殊的地位; 特性列表 多种语言和协议编写客户端...

次渠龙哥
2018/06/26
0
0
用java代码如何设置activemq消息持久化到数据库中?

想将activeMQ消息持久化到数据库中,网上都是说将activeMQ.xml文件做如下设置: 我能否在java代码中进行设置呢?因为我是在eclipse中导入的activemq-all-5.2.0.jar包,开发的系统,包中没有a...

吕奕
2010/03/20
2.4K
0

没有更多内容

加载失败,请刷新页面

加载更多

好程序员大数据教程分享Scala系列之模式匹配和样例类

好程序员大数据教程分享Scala系列之模式匹配和样例类 1.样例类 在Scala中样例类是一中特殊的类,样例类是不可变的, 可以通过值进行比较,可用于模式匹配。 定义一个样例类: 构造器中每一个...

好程序员官网
11分钟前
3
0
让nginx上的静态网页在访问的时候没有html后缀

需求背景: 公司产品小姐姐觉得这个访问带html后缀不专业,要求访问不带html后缀 nginx 配置 #原配置 location / { index index.html index.htm index.php; try_files $...

Linux_Anna
12分钟前
2
0
beetl的内置函数

函数调用Beetl内置函数请参考附录,以下列出了常用的函数date 返回一个java.util.Date类型的变量,如 date() 返回一个当前时间(对应java的java.util.Date); ${date( "2011-1-1" , "yy...

gantaos
12分钟前
3
0
spring cloud 2.x版本 Gateway自定义过滤器教程

前言 本文采用Spring cloud本文为2.1.8RELEASE,version=Greenwich.SR3 [toc] 本文基于前两篇文章eureka-server、eureka-client、eureka-ribbon、eureka-feign和spring-gataway的实现。 参考......

毛毛向前冲V5
16分钟前
3
0
VPGAME 的 Kubernetes 迁移实践

作者 | 伍冲斌 VPGAME 运维开发工程师 导读:VPGAME 是集赛事运营、媒体资讯、大数据分析、玩家社群、游戏周边等为一体的综合电竞服务平台。总部位于中国杭州,在上海和美国西雅图分别设立了...

阿里巴巴云原生
21分钟前
3
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部