当前位置：移动技术网 > IT编程>开发语言>Java > JMS简介与ActiveMQ实战代码分享

JMS简介与ActiveMQ实战代码分享

2019年07月19日 | 移动技术网IT编程 | 我要评论

一、异步通信

之前接触到的rmi，hessian等技术都是同步通信机制。当客户端调用远程方法时，客户端必须等到远程方法完成后，才能继续执行。这段时间客户端一直会被阻塞（这样造成的用户体验很不好）。

（同步通信）

同步通信有并不是程序之间交互的唯一方式，异步通信机制中，客户端不需要等待服务处理消息，可以继续执行，并且最终能够收到并处理消息。

（异步通信）

异步通信的优势

无需等待。客户端只需要将消息发送给消息代理，不需要等待就可以继续执行别的任务，且确信消息会被投递给相应的目的地。
面向消息和解耦。客户端不需要担心远程服务的接口规范，只需要把消息放入消息队列然后获取结果即可。

二、jms

1. 简介

在jms出现之前，每个消息代理都是有不同的实现，这就使得不同代理之间的消息代码很难通用。jms（java message service，java消息服务）是一个标准，定义了使用消息代理的通用api。即所有遵从规范的实现都使用通用的接口，类似于jdbc为数据库操作提供通用接口。

jms几个重要的要素：

destination：消息从发送端发出后要走的通道。
connectionfactory：连接工厂，用于创建连接的对象。
connection：连接接口，用于创建session。
session：会话接口，用于创建消息的发送者，接受者以及消息对象本身。
messageconsumer：消息的消费者。
messageproducer：消息的生产者。
xxxmessage：各种类型的消息对象，包括bytemessage、mapmessage、objectmessage、streammessage和textmessage 5种。

2. jms消息模型

不同的消息系统有不同的消息模型。jms提供了两种模型：queue（点对点）和topic（发布/订阅）。

jms queue（点对点）模型

在点对点模型中，消息生产者生产消息发送到queue中，然后消息消费者从queue中取出并且消费消息，但不可重复消费。

如图：

发送者1，发送者2，发送者3各发送一条消息到服务器;

消息1，2，3就会按照顺序形成一个队列，队列中的消息不知道自己会被哪个接收者消费;

接收者1，2，3分别从队列中取出一条消息进行消费，每取出一条消息，队列就会将该消息删除，这样即保证了消息不会被重复消费。

jms queue模型也成为p2p(point to point)模型。

jms topic(发布/订阅)模型

jms topic模型与jms queue模型的最大差别在于消息接收的部分。topic模型类似于微信公众号，订阅了该公众号的接收者都可以接收到公众号推送的消息。

如图：

发布者1，2，3分别发布3个主题1，2，3；
这样订阅了主题1的用户群：订阅者1，2，3即能接收到主题1消息；同理订阅者4，5，6即能接收到主题2消息，订阅者7，8，9即能接收到主题3消息。

jms topic模型也成为pus/sub模型。

两种模式下各要素的对比：

3. 传统jms编程模型

producer：

（1）创建连接工厂connectionfactory；
（2）使用连接工厂创建连接；
（3）启动连接；
（4）创建会话；
（5）创建消息发送的目的地；
（6）创建生产者；
（7）创建消息类型和消息内容；
（8）发送消息；

consumer：

（1）创建连接工厂connectionfactory；
（2）使用连接工厂创建连接；
（3）启动连接；
（4）创建会话；
（5）创建消息发送的目的地；
（6）创建消费者
（7）创建消息类型；
（8）接收消息；

三、 activemq简介

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

activemq 主要特性：

多种语言和协议编写客户端。语言: java,c,c++,c#,ruby,perl,python,php。应用协议：
openwire,stomp rest,ws notification,xmpp,amqp
完全支持jms1.1和j2ee 1.4规范（持久化，xa消息，事务)
对spring的支持，activemq可以很容易内嵌到使用spring的系统里面去，而且也支持spring2.0的特性
通过了常见j2ee服务器（如 geronimo,jboss 4,glassfish,weblogic)的测试，其中通过jca 1.5 resource adaptors的配置，可以让activemq可以自动的部署到任何兼容j2ee 1.4 商业服务器上
支持多种传送协议：in-vm,tcp,ssl,nio,udp,jgroups,jxta
支持通过jdbc和journal提供高速的消息持久化
从设计上保证了高性能的集群，客户端-服务器，点对点
支持ajax
支持与axis的整合
可以很容易得调用内嵌jms provider，进行测试

四、 activemq实战

下面看看如何activemq实现一个简单的消息队列。

传统的jms编程模型

1. jms queue模型代码实现：

producer：

package com.wgs.mq.queue;
import org.apache.activemq.activemqconnectionfactory;
import javax.jms.*;
/**
 * created by genshenwang.nomico on 2017/10/19.
 */
public class activemqproducer {
	private static final string url = "tcp://localhost:61616";
	private static final string queue_name = "queue-name";
	public static void main(string[] args) throws jmsexception {
		//1 创建连接工厂connectionfactory
		connectionfactory connectionfactory = new activemqconnectionfactory(url);
		//2 使用连接工厂创建连接
		connection connection = connectionfactory.createconnection();
		//3 启动连接
		connection.start();
		//4 创建会话
		session session = connection.createsession(false, session.auto_acknowledge);
		//5 创建消息发送的目的地
		destination destination = session.createqueue(queue_name);
		//6 创建生产者
		messageproducer messageproducer = session.createproducer(destination);
		//7 创建消息
		textmessage textmessage = session.createtextmessage();
		for (int i = 1; i <= 100; i++) {
			//8 创建消息内容
			textmessage.settext("发送者- 1 -发送消息：" + i);
			//9 发送消息
			messageproducer.send(textmessage);
		}
		system.out.println("消息发送成功");
		session.close();
		connection.close();
	}
}

conusmer：

package com.wgs.mq.queue;
import org.apache.activemq.activemqconnectionfactory;
import javax.jms.*;
/**
 * created by genshenwang.nomico on 2017/10/19.
 */
public class activemqconsumer {
	private static final string url = "tcp://localhost:61616";
	private static final string queue_name = "queue-name";
	public static void main(string[] args) throws jmsexception {
		//1 创建连接工厂connectionfactory
		connectionfactory connectionfactory = new activemqconnectionfactory(url);
		//2 使用连接工厂创建连接
		connection connection = connectionfactory.createconnection();
		//3 启动连接
		connection.start();
		//4 创建会话
		session session = connection.createsession(false, session.auto_acknowledge);
		//5 创建消息发送的目的地
		destination destination = session.createqueue(queue_name);
		//6 创建消费者
		messageconsumer messageconsumer = session.createconsumer(destination);
		messageconsumer.setmessagelistener(new messagelistener() {
			public void onmessage(message message) {
				//7 创建消息
				textmessage textmessage = (textmessage)message;
				try {
					//7 接收消息
					system.out.println("消费者- 1 -接收消息：【" + textmessage.gettext() + "】");
				}
				catch (jmsexception e) {
					e.printstacktrace();
				}
			}
		}
		);
	}
}

2. jms topic模型代码实现：

producer：

package com.wgs.mq.topic;
import org.apache.activemq.activemqconnectionfactory;
import javax.jms.*;
/**
 * 发布订阅模式
 * created by genshenwang.nomico on 2017/10/19.
 */
public class activemqproducer {
	private static final string url = "tcp://localhost:61616";
	private static final string topic_name = "topic-name";
	public static void main(string[] args) throws jmsexception {
		//1 创建连接工厂connectionfactory
		connectionfactory connectionfactory = new activemqconnectionfactory(url);
		//2 使用连接工厂创建连接
		connection connection = connectionfactory.createconnection();
		//3 启动连接
		connection.start();
		//4 创建会话
		session session = connection.createsession(false, session.auto_acknowledge);
		//5 创建带有主题的消息发送的目的地
		destination destination = session.createtopic(topic_name);
		//6 创建生产者
		messageproducer messageproducer = session.createproducer(destination);
		//7 创建消息
		textmessage textmessage = session.createtextmessage();
		for (int i = 1; i <= 100; i++) {
			//8 创建消息内容
			textmessage.settext("发送者- 1 -发送消息：" + i);
			//9 发送消息
			messageproducer.send(textmessage);
		}
		system.out.println("消息发送成功");
		session.close();
		connection.close();
	}
}

consumer：

package com.wgs.mq.topic;
import org.apache.activemq.activemqconnectionfactory;
import javax.jms.*;
/**
 * 发布订阅模式
 * created by genshenwang.nomico on 2017/10/19.
 */
public class activemqconsumer {
	private static final string url = "tcp://localhost:61616";
	private static final string topic_name = "topic-name";
	public static void main(string[] args) throws jmsexception {
		//1 创建连接工厂connectionfactory
		connectionfactory connectionfactory = new activemqconnectionfactory(url);
		//2 使用连接工厂创建连接
		connection connection = connectionfactory.createconnection();
		//3 启动连接
		connection.start();
		//4 创建会话
		session session = connection.createsession(false, session.auto_acknowledge);
		//5 创建消息发送的目的地
		destination destination = session.createtopic(topic_name);
		//6 创建消费者
		messageconsumer messageconsumer = session.createconsumer(destination);
		messageconsumer.setmessagelistener(new messagelistener() {
			public void onmessage(message message) {
				//7 创建消息
				textmessage textmessage = (textmessage)message;
				try {
					//7 接收消息
					system.out.println("消费者- 1 -接收消息：【" + textmessage.gettext() + "】");
				}
				catch (jmsexception e) {
					e.printstacktrace();
				}
			}
		}
		);
	}
}

使用spring的jms模板

虽然jms为所有的消息代理提供了统一的接口，但如同jdbc一样，在处理连接，语句，结果集和异常时会显得很繁杂。不过，spring为我们提供了jmstemplate来消除冗余和重复的jms代码。
下面看看如何使用jmstemplate来实现消息队列。

1. jms queue模型代码实现：

配置文件：
producer.xml:

<?xml version="1.0" encoding="utf-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/xmlschema-instance"
    xmlns:context="http://www.springframework.org/schema/context"
    xsi:schemalocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">

  <context:annotation-config/>

  <!-- activemq提供的connectionfactory-->
  <bean id="targetconnectionfactory" class="org.apache.activemq.activemqconnectionfactory">
    <property name="brokerurl" value="tcp://localhost:61616"/>
  </bean>

  <!-- 在spring 中配置jms连接工厂，连接到activemq提供的connectionfactory-->
  <bean id="connectionfactory" class="org.springframework.jms.connection.singleconnectionfactory">
    <property name="targetconnectionfactory" ref = "targetconnectionfactory"/>
  </bean>

  <!-- 配置jmstemplate，用于发送消息 -->
  <bean id="jmstemplate" class="org.springframework.jms.core.jmstemplate">
    <property name="connectionfactory" ref="connectionfactory"/>
  </bean>

  <!-- 配置队列目的地的名称-->
  <bean id="queuedestination" class="org.apache.activemq.command.activemqqueue">
    <constructor-arg value="queue-spring-name"/>
  </bean>
  <!-- 配置队列目的地的名称-->
  <bean id="topicdestination" class="org.apache.activemq.command.activemqtopic">
    <constructor-arg value="topic-spring-name"/>
  </bean>

  <bean id="producerserviceimpl" class="com.wgs.jms.producer.activemqproducerserviceimpl"/>


</beans>

consumer.xml:

<?xml version="1.0" encoding="utf-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/xmlschema-instance"
    xmlns:context="http://www.springframework.org/schema/context"
    xsi:schemalocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">

  <context:annotation-config/>

  <!-- activemq提供的connectionfactory-->
  <bean id="targetconnectionfactory" class="org.apache.activemq.activemqconnectionfactory">
    <property name="brokerurl" value="tcp://localhost:61616"/>
  </bean>

  <!-- 在spring 中配置jms连接工厂，连接到activemq提供的connectionfactory-->
  <bean id="connectionfactory" class="org.springframework.jms.connection.singleconnectionfactory">
    <property name="targetconnectionfactory" ref = "targetconnectionfactory"/>
  </bean>

  <!-- 配置队列目的地的名称-->
  <bean id="queuedestination" class="org.apache.activemq.command.activemqqueue">
    <constructor-arg value="queue-spring-name"/>
  </bean>
  <!-- 配置消息监听器-->
  <bean id="consumermessagelistener" class="com.wgs.jms.consumer.consumermessagelistener"/>
  <!-- 配置队列目的地的名称-->
  <bean id="jmscontainer" class="org.springframework.jms.listener.defaultmessagelistenercontainer">
    <property name="destination" ref="queuedestination"/>
    <property name="connectionfactory" ref="connectionfactory"/>
    <property name="messagelistener" ref="consumermessagelistener"/>
  </bean>
  <!-- 配置队列目的地的名称-->
  <bean id="topicdestination" class="org.apache.activemq.command.activemqtopic">
    <constructor-arg value="topic-spring-name"/>
  </bean>
</beans>

生产者producer：

（1）先写一个接口：

package com.wgs.jms.producer;
/**
 * created by genshenwang.nomico on 2017/10/20.
 */
public interface activemqproducerservice {
  void sendmessage(final string message);
}

（2）接口的实现：

package com.wgs.jms.producer;
import org.springframework.beans.factory.annotation.autowired;
import org.springframework.jms.core.jmstemplate;
import org.springframework.jms.core.messagecreator;
import javax.annotation.resource;
import javax.jms.*;
/**
 * created by genshenwang.nomico on 2017/10/20.
 */
public class activemqproducerserviceimpl implements activemqproducerservice {
	@autowired
	  jmstemplate jmstemplate;
	@resource(name = "queuedestination")
	  destination destination;
	public void sendmessage(final string message) {
		jmstemplate.send(destination, new messagecreator() {
			public message createmessage(session session) throws jmsexception {
				textmessage textmessage = session.createtextmessage(message);
				return textmessage;
			}
		}
		);
		system.out.println("生产者- 1 -发送消息成功：" + message);
	}
}

（3）测试：

package com.wgs.jms.producer;
import org.springframework.context.support.classpathxmlapplicationcontext;
/**
 * created by genshenwang.nomico on 2017/10/20.
 */
public class activemqproducermain {
	public static void main(string[] args) {
		classpathxmlapplicationcontext context = new classpathxmlapplicationcontext("producer.xml");
		activemqproducerservice service = context.getbean(activemqproducerservice.class);
		for (int i = 0; i < 100; i++) {
			service.sendmessage("test" + i);
		}
		context.close();
	}
}

消费者：

（1）创建消息监听器：

package com.wgs.jms.consumer;
import javax.jms.jmsexception;
import javax.jms.message;
import javax.jms.messagelistener;
import javax.jms.textmessage;
/**
 * created by genshenwang.nomico on 2017/10/20.
 */
public class consumermessagelistener implements messagelistener {
	public void onmessage(message message) {
		try {
			textmessage textmessage = (textmessage) message;
			system.out.println("消费者- 1 -接收消息：" + textmessage.gettext());
		}
		catch (jmsexception e) {
			e.printstacktrace();
		}
	}
}

（2）测试：

package com.wgs.jms.consumer;
import org.springframework.context.support.classpathxmlapplicationcontext;
/**
 * created by genshenwang.nomico on 2017/10/20.
 */
public class activemqconsumermain {
	public static void main(string[] args) {
		classpathxmlapplicationcontext context = new classpathxmlapplicationcontext("consumer.xml");
	}
}

2. jms topic模型代码实现：

将上述代码中出现的queuedestination改为topicdestination即可。

总结

以上就是本文关于jms简介与activemq实战代码分享的全部内容，希望对大家有所帮助。感兴趣的朋友可以继续参阅本站其他相关专题，如有不足之处，欢迎留言指出。感谢朋友们对本站的支持！

您可能感兴趣的文章:

如对本文有疑问，点击进行留言回复！！

POJ-3279 枚举+dfs

由于上一行的状态决定了下一行的翻转，所以只需要枚举第一行的翻转情况（用二进制状态下1表示翻转），后面的行就能确... [阅读全文]
Spring学习笔记-IOC容器

IOC容器学习笔记什么是IOC：控制反转，把对象创建和对象之间的调用过程，交给Spring进行管理。使用IOC是... [阅读全文]
java Map及其实现类的底层原理

文章目录一、Map接口及其多个实现类的对比二、Map中存储的key-value特点三、HashMap在JDK7中... [阅读全文]
java 算法练习1

输入一个数，计算所有的位数之和import java.util.Scanner;public class sum... [阅读全文]
集合框架——Map、泛型以及Collection算法常用方法

MapMap接口：存储一组键值对象，提供key到value的映射Map接口专门处理键值映射数据的存储，可以根据键... [阅读全文]
Elasticsearch 升级 7.x 版本后，我感觉掉坑里了

最近想把我的mall项目升级下，支持SpringBoot 2.3.0 版本。升级过程中发现需要升级Elastic... [阅读全文]
Mybatis的插件运行原理以及如何编写一个Mybatis的插件

Mybatis作为一个优秀的ORM插件有很强大的灵活性，通过插件可以很方便地扩展Mybatis的功能。Mybat... [阅读全文]
数据结构和算法 - 插入排序

基本介绍把 n 个待排序的元素看成一个有序表和一个无序表，开始时有序表只有一个值，无序表有 n-1 个值，每次排... [阅读全文]
IntelliJ IDEA 2020.2 简单配置

1、Idea 设置字体settings --> Editor --> Font2、配置MavenSe... [阅读全文]
变量，常量，数据类型，运算符（Java语言）

今天我们来简单聊聊变量，常量，数据类型，运算符的一些东西，希望可以帮到大家。常量常量很简单：通俗的说就是代码编译... [阅读全文]

网友评论


验证码：

JMS简介与ActiveMQ实战代码分享

2019年07月19日 | 移动技术网IT编程 | 我要评论

您可能感兴趣的文章:

相关文章:

网友评论