消息队列（MQ）是一种应用程序对应用程序的通信方法，应用程序通过队列进行通信，而不是通过直接调用彼此来通信，队列的使用除去了接收和发送应用程序同时执行的要求。是进行通信的中间件产品。（换言之：MQ负责两个系统之间传递消息，这两个系统可以是异构的，处于不同硬件、不同操作系统、用不同语言编写，只需要简单的调用几个MQ的API，就可以互相通讯，你不必考虑底层系统和网络的复杂性。MQ能够应付多种异常情况，例如网络阻塞、临时中断等等）
PS：直接调用通常是用于诸如远程过程调用的技术。
补充知识：MB(消息路由、数据转换)
优秀MQ特点
　　a&.高可用性，希望MQ能支撑7x24小时应用，而不是三天两头当机，要做到高可用性，就需要做MQ的集群，一台当了，不影响整个集群的服务能力，这里涉及到告警、流控、消息的负载均衡、数据库的使用、测试的完备程度等等。
　　b&.消息存储的高可靠性。要保证100%不丢消息。这不仅仅是MQ的责任，更涉及到硬件、操作系统、语言平台和数据库的一整套方案。许多号称可靠存储的MQ产品其实都不可靠，要知道，硬件错误是常态，如果在硬件错误的情况下还能保证消息的可靠存储这才是难题。这里可能需要用到特殊的存储硬件，特殊的数据库，分布式的数据存储，数据库的分库分表和同步等等。你要考虑消息存储在哪里，是文件系统，还是数据库，是本地文件，还是分布式文 件，是搞主辅备份呢还是多主机写入等等。
　　c&.高可扩展性，MQ集群能很好地支持水平扩展，这就要求我们的节点之间最好不要有通信和数据同步。
　　d&.性能，性能是实现高可用性的前提，很难想象单机性能极差的MQ组成的集群能在高负载下幸免于难。性能因素跟采用的平台、语言、操作系统、代码质量、数据 库、网络息息相关。MQ产品的核心其实是消息的存储，在保证存储安全的前提下如何保证和提高消息入队的效率是性能的关键因素。这里需要开发人员建立起性能观念，不需要你对一行行代码斤斤计较，但是你需要知道这样做会造成什么后果，有没有更好更快的方式，你怎么证明它更好更快。软件实现的性能是一方面，另一 方面就是平台相关了，因为MQ本质上是IO密集型的系统，瓶颈在IO，如何优化网络IO、文件IO这需要专门的知识。性能另一个相关因素是消息的调度上， 引入消息顺序和消息优先级，允许消息的延迟发送，都将增大消息发送调度的复杂性，如何保证高负载下的调度也是要特别注意的地方。
　　e&.高可配置性和监控工具的完整，这是一个MQ产品容易忽略的地方。异步通信造成了查找问题的难度，不像同步调用那样有相对明确的时序关系。因此查找异步通信 的异常是很困难的，这就需要MQ提供方便的DEBUG工具，查找分析日志的工具，查看消息生命周期的工具，查看系统间依赖关系的工具等等。可定制也是MQ 产品非常重要的一方面，可方便地配置各类参数并在集群中同步，并且可动态调整各类参数，这将大大降低维护难度。
Ps:一句话总结：全天候不宕机，安全消息存储，100%不丢失数据，高效率的写入读出，同时要求方便查错。
产品比较　
是使用Erlang编写的一个开源的消息队列，本身支持很多的协议：AMQP，XMPP, SMTP, STOMP，也正是如此，使的它变的非常重量级，更适合于企业级的开发。同时实现了一个经纪人(Broker)构架，这意味着消息在发送给客户端时先在中 心队列排队。对路由(Routing)，负载均衡(Load balance)或者数据持久化都有很好的支持。
Ps: 结合erlang语言本身的并发优势，性能较好，但是不利于做二次开发和维护
是一个Key-Value的NoSQL数据库，开发维护很活跃，虽然它是一个Key-Value数据库存储系统，但它本身支持MQ功能，所以完全可 以当做一个轻量级的队列服务来使用。对于RabbitMQ和Redis的入队和出队操作，各执行100万次，每10万次记录一次执行时间。测试数据分为 128Bytes、512Bytes、1K和10K四个不同大小的数据。实验表明：入队时，当数据比较小时Redis的性能要高于RabbitMQ，而如 果数据大小超过了10K，Redis则慢的无法忍受；出队时，无论数据大小，Redis都表现出非常好的性能，而RabbitMQ的出队性能则远低于 Redis。
&&&&&&&&&&&&&&&&&&&& 入队
&&&&&&&&&&&&&&&&&&&& 出队
Ps: 做为一个基于内存的K-V数据库，其提供了消息订阅的服务，可以当作MQ来使用，目前应用案例较少，且不方便扩展
号称最快的消息队列系统，尤其针对大吞吐量的需求场景。ZMQ能够实现RabbitMQ不擅长的高级/复杂的队列，但是开发人员需要自己组合多种技 术框架，技术上的复杂度是对这MQ能够应用成功的挑战。ZeroMQ具有一个独特的非中间件的模式，你不需要安装和运行一个消息服务器或中间件，因为你的 应用程序将扮演了这个服务角色。你只需要简单的引用ZeroMQ程序库，可以使用NuGet安装，然后你就可以愉快的在应用程序之间发送消息了。但是 ZeroMQ仅提供非持久性的队列，也就是说如果down机，数据将会丢失。其中，Twitter的Storm中使用ZeroMQ作为数据流的传输。
Ps: 扩展性好，开发比较灵活，采用C语言实现，实际上他只是一个socket库的重新封装，如果我们做为消息队列使用，需要开发大量的代码
是Apache下的一个子项目。 类似于ZeroMQ，它能够以代理人和点对点的技术实现队列。同时类似于RabbitMQ，它少量代码就可以高效地实现高级应用场景。RabbitMQ、 ZeroMQ、ActiveMQ均支持常用的多种语言客户端 C++、Java、.Net,、Python、 Php、 Ruby等。
Ps: 历史悠久的开源项目，已经在很多产品中得到应用，实现了JMS1.1规范，可以和spring-jms轻松融合，实现了多种协议，不够轻巧（源代码比RocketMQ多）.支持持久化到数据库，对队列数较多的情况支持不好.
Jafka/Kafka
Kafka是Apache下的一个子项目，是一个高性能跨语言分布式Publish/Subscribe消息队列系统，而Jafka是在Kafka 之上孵化而来的，即Kafka的一个升级版。具有以下特性：快速持久化，可以在O(1)的系统开销下进行消息持久化；高吞吐，在一台普通的服务器上既可以 达到10W/s的吞吐速率；完全的分布式系统，Broker、Producer、Consumer都原生自动支持分布式，自动实现复杂均衡；支持 Hadoop数据并行加载，对于像Hadoop的一样的日志数据和离线分析系统，但又要求实时处理的限制，这是一个可行的解决方案。Kafka通过 Hadoop的并行加载机制来统一了在线和离线的消息处理，这一点也是本课题所研究系统所看重的。Apache Kafka相对于ActiveMQ是一个非常轻量级的消息系统，除了性能非常好之外，还是一个工作良好的分布式系统。
阿里巴巴的MQ中间件，在其多个产品下使用，并能够撑住双十一的大流量，他并没有实现JMS规范，使用起来很简单。部署由一个&命名服务（nameserver）和一个代理（broker）组成，nameserver和broker以及producer都支持集群，队列的容量受机器硬盘的限制，队列满后可以支持持久化到硬盘（也可以自己适配代码，将其持久化到NOSQL数据库中），队列满后会影响吞吐量，可以采用主备来保证稳定性，支持回溯消费，可以在broker端进行消息过滤.
其他一些队列列表HornetQ、Apache Qpid、Sparrow、Starling、Kestrel、Beanstalkd、Amazon SQS就不再一一分析。
比较ActiveMQ and RocketMQ
需要新建一个特殊队列来接收优先级高的队列，无法实现从0-65535这种细粒度的控制
可以精细控制
可以保证严格的消费顺序
无法保证严格的顺序
RocketMQ可以在broker端进行过滤，对于我们的消息总线，这里可以节省大量的网络传输是否会有消息重发造成的重复消费：RocketMQ可以保证，ActiveMQ无法保证
仅支持在客户端消费的时候进行判断是否是自己需要的消息
即重新将某一个时刻之前的消息重新消费一遍
就是当缓存消息的内存满了之后的解决方案，一种是丢弃策略，这种不会影响吞吐量，还有一种就是将消息持久化到磁盘，这种会影响吞吐量
客户端不在线
RocketMQ可以在客户端上线后继续将未消费的消息推送到客户端
比较主流的MQ：
所属社区/公司
&Mozilla Public License
社区活跃度&&
功能齐全，被大量开源项目使用
由于Erlang&语言的并发能力，性能很好&&
&各个环节分布式扩展设计，主从&HA；支持上万个队列；多种消费模式；性能很好
低延时，高性能，最高&43万条消息每秒&&
授权方式&&
开发语言&&
支持的协议&&
OpenWire、STOMP、REST、XMPP、AMQP
自己定义的一套(社区提供JMS--不成熟)
客户端支持语言&&
Java、C、C++、Python、PHP、Perl、.net&等&
Java、C、C++、Python、&PHP、Perl&等
Java、C++（不成熟）&&
python、&java、&php、.net&等
内存、文件、数据库
内存、文件
在消息发送端保存
管理界面&&
无社区有&webconsole&&&实现
部署方式&&
独立、嵌入
优点：成熟的产品，已经在很多公司得到应用（非大规模场景）。有较多的文档。各种协议支持较好，有多重语言的成熟的客户端；
缺点：根据其他用户反馈，会出莫名其妙的问题，切会丢失消息。其重心放到activemq6.0&产品&apollo上去了，目前社区不活跃，且对5.x维护较少；
Activemq不适合用于上千个队列的应用场景
优点：由于erlang语言的特性，mq性能较好；管理界面较丰富，在互联网公司也有较大规模的应用；支持amqp系诶，有多中语言且支持amqp的客户端可用
缺点：erlang语言难度较大。集群不支持动态扩展。
优点：模型简单，接口易用（JMS的接口很多场合并不太实用）。在阿里大规模应用。目前支付宝中的余额宝等新兴产品均使用rocketmq。集群规模大概在50&台左右，单日处理消息上百亿；性能非常好，可以大量堆积消息在broker中；支持多种消费，包括集群消费、广播消费等。开发度较活跃，版本更新很快。
缺点：没有在mq核心中去实现JMS等接口，
&实例（简单的实战）
&　　ActiveMQ入门实例：
　　　　1.去官方网站下载： 　　　　2.运行ActiveMQ
解压缩apache-activemq-5.8.0-bin.zip，然后双击apache-activemq-5.5.1\bin\activemq.bat运行ActiveMQ程序。
　　　　　　启动ActiveMQ以后，登陆：http://localhost:8161/admin/，创建一个Queue，命名为FirstQueue。
　　　　3.创建Eclipse项目并运行
　　　　创建java project：ActiveMQ，新建lib文件夹,导入如下Jar包
　　　　activemq-broker-5.8.0.jar 、activemq-client-5.8.0.jar
、geronimo-j2ee-management_1.1_spec-1.0.1.jar
、geronimo-jms_1.1_spec-1.1.1.jar
、slf4j-api-1.6.6.jar
　　　　创建类如下类：　　　　
import javax.jms.C
import javax.jms.ConnectionF
import javax.jms.DeliveryM
import javax.jms.D
import javax.jms.MessageP
import javax.jms.S
import javax.jms.TextM
import org.apache.activemq.ActiveMQC
import org.apache.activemq.ActiveMQConnectionF
public class Sender {
private static final int SEND_NUMBER = 5;
public static void main(String[] args) {
// ConnectionFactory ：连接工厂，JMS 用它创建连接
ConnectionFactory connectionF // Connection ：JMS 客户端到JMS
// Provider 的连接
Connection connection = null; // Session： 一个发送或接收消息的线程
S // Destination ：消息的目的地;消息发送给谁.
Des // MessageProducer：消息发送者
MessageP // TextM
// 构造ConnectionFactory实例对象，此处采用ActiveMq的实现jar
connectionFactory = new ActiveMQConnectionFactory(
ActiveMQConnection.DEFAULT_USER,
ActiveMQConnection.DEFAULT_PASSWORD, "tcp://localhost:61616");
try { // 构造从工厂得到连接对象
connection = connectionFactory.createConnection();
connection.start();
// 获取操作连接
session = connection.createSession(Boolean.TRUE,
Session.AUTO_ACKNOWLEDGE);
// 获取session注意参数值xingbo.xu-queue是一个服务器的queue，须在在ActiveMq的console配置
destination = session.createQueue("FirstQueue");
// 得到消息生成者【发送者】
producer = session.createProducer(destination);
// 设置不持久化，此处学习，实际根据项目决定
producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
// 构造消息，此处写死，项目就是参数，或者方法获取
sendMessage(session, producer);
} catch (Exception e) {
e.printStackTrace();
} finally {
if (null != connection)
connection.close();
} catch (Throwable ignore) {
public static void sendMessage(Session session, MessageProducer producer)
throws Exception {
for (int i = 1; i &= SEND_NUMBER; i++) {
TextMessage message = session.createTextMessage("ActiveMq 发送的消息"
// 发送消息到目的地方
System.out.println("发送消息：" + "ActiveMq 发送的消息" + i);
producer.send(message);
import javax.jms.C
import javax.jms.ConnectionF
import javax.jms.D
import javax.jms.MessageC
import javax.jms.S
import javax.jms.TextM
import org.apache.activemq.ActiveMQC
import org.apache.activemq.ActiveMQConnectionF
public class Receiver {
public static void main(String[] args) {
// ConnectionFactory ：连接工厂，JMS 用它创建连接
ConnectionFactory connectionF
// Connection ：JMS 客户端到JMS Provider 的连接
Connection connection = null;
// Session： 一个发送或接收消息的线程
// Destination ：消息的目的地;消息发送给谁.
// 消费者，消息接收者
connectionFactory = new ActiveMQConnectionFactory(
ActiveMQConnection.DEFAULT_USER,
ActiveMQConnection.DEFAULT_PASSWORD, "tcp://localhost:61616");
// 构造从工厂得到连接对象
connection = connectionFactory.createConnection();
connection.start();
// 获取操作连接
session = connection.createSession(Boolean.FALSE,
Session.AUTO_ACKNOWLEDGE);
// 获取session注意参数值xingbo.xu-queue是一个服务器的queue，须在在ActiveMq的console配置
destination = session.createQueue("FirstQueue");
consumer = session.createConsumer(destination);
while (true) {
// 设置接收者接收消息的时间，为了便于测试，这里谁定为100s
TextMessage message = (TextMessage) consumer.receive(100000);
if (null != message) {
System.out.println("收到消息" + message.getText());
} catch (Exception e) {
e.printStackTrace();
} finally {
if (null != connection)
connection.close();
} catch (Throwable ignore) {
关于消息队列与分布式的那些事
　　消息队列技术是分布式应用间交换信息的一种技术。消息队列可驻留在内存或磁盘上,队列存储消息直到它们被应用程序读走。通过消息队列，应用程序可独立地执行--它们不需要知道彼此的位置、或在继续执行前不需要等待接收程序接收此消息。在分布式计算环境中，为了集成分布式应用，开发者需要对异构网络环境下的分布式应用提供有效的通信手段。为了管理需要共享的信息，对应用提供公共的信息交换机制是重要的。
设计分布式应用的方法主要有：
远程过程调用(PRC)--分布式计算环境(DCE)的基础标准成分之一；
对象事务监控(OTM)--基于CORBA的面向对象工业标准与事务处理(TP)监控技术的组合；
消息队列(MessageQueue)--构造分布式应用的松耦合方法。
　　(a) 分布计算环境/远程过程调用 (DCE/RPC)
　　RPC是DCE的成分，是一个由开放软件基金会(OSF)发布的应用集成的软件标准。RPC模仿一个程序用函数引用来引用另一程序的传统程序设计方法，此引用是过程调用的形式，一旦被调用，程序的控制则　　　　转向被调用程序。
　　在RPC实现时，被调用过程可在本地或远地的另一系统中驻留并在执行。当被调用程序完成处理输入数据，结果放在过程调用的返回变量中返回到调用程序。RPC完成后程序控制则立即返回到调用程序。因此　　　　RPC模仿子程序的调用/返回结构，它仅提供了Client(调用程序)和Server(被调用过程)间的同步数据交换。
　　(b) 对象事务监控 (OTM)
　　基于CORBA的面向对象工业标准与事务处理(TP)监控技术的组合，在CORBA规范中定义了：使用面向对象技术和方法的体系结构；公共的Client/Server程序设计接口；多平台间传输和翻译数据的指导方　　　　　针；开发分布式应用接口的语言(IDL)等，并为构造分布的Client/Server应用提供了广泛及一致的模式。
　　(c) 消息队列 (Message Queue)
　　消息队列为构造以同步或异步方式实现的分布式应用提供了松耦合方法。消息队列的API调用被嵌入到新的或现存的应用中，通过消息发送到内存或基于磁盘的队列或从它读出而提供信息交换。消息队列可用　　　　在应用中以执行多种功能，比如要求服务、交换信息或异步处理等。
中间件是一种独立的系统软件或服务程序，分布式应用系统借助这种软件在不同的技术之间共享资源，管理计算资源和网络通讯。它在计算机系统中是一个关键软件，它能实现应用的互连和互操作性，能保证　　　　系统的安全、可靠、高效的运行。中间件位于用户应用和操作系统及网络软件之间，它为应用提供了公用的通信手段，并且独立于网络和操作系统。中间件为开发者提供了公用于所有环境的应用程序接口，当　　　　应用程序中嵌入其函数调用，它便可利用其运行的特定操作系统和网络环境的功能，为应用执行通信功能。
如果没有消息中间件完成信息交换，应用开发者为了传输数据，必须要学会如何用网络和操作系统软件的功能，编写相应的应用程序来发送和接收信息，且交换信息没有标准方法，每个应用必须进行特定的编程从而和多平台、不同环境下的一个或多个应用通信。例如，为了实现网络上不同主机系统间的通信，将要求具备在网络上如何交换信息的知识（比如用TCP/IP的socket程序设计）；为了实现同一主机内不同进程之间的通讯，将要求具备操作系统的消息队列或命名管道(Pipes)等知识。
　　目前中间件的种类很多，如交易管理中间件(如IBM的CICS)、面向Java应用的Web应用服务器中间件(如IBM的WebSphere Application Server)等，而消息传输中间件(MOM)是其中的一种。它简化了应用之间数据的传输，屏蔽底层异构操作系统和网络平台，提供一致的通讯标准和应用开发，确保分布式计算网络环境下可靠的、跨平台的信息传输和数据交换。它基于消息队列的存储-转发机制，并提供特有的异步传输机制，能够基于消息传输和异步事务处理实现应用整合与数据交换。
阅读(...) 评论()JMS之ActiveMQ Linux下安装与应用实例_Linux教程_Linux公社-Linux系统门户网站
你好，游客
JMS之ActiveMQ Linux下安装与应用实例
来源：Linux社区&
作者：wuzhaohuixy
JMS之ActiveMQ Linux下安装与应用实例
1.下载activeMQ安装包,拷贝到/activeMQ目录下
apache-activemq-5.10.0-bin.tar.gz，下载地址
2.解压文件到运行目录
[root@linuxidc softs]# tar -xzvf /server/apache-activemq-5.10.0-bin.tar.gz
3.为了方便管理，重命名
[root@linuxidc softs]# mv apache-activemq-5.10.0 activemq-5.10.0
[root@linuxidc softs]# cd activemq-5.10.0/[root@linuxidc activemq-5.10.0]# lltotal 6304-rwxr-xr-x 1 root root 6371237 Jun& 5& 2014 activemq-all-5.10.0.jardrwxr-xr-x 5 root root& & 4096 Jan 11 23:31 bindrwxr-xr-x 2 root root& & 4096 Jan 11 23:31 confdrwxr-xr-x 2 root root& & 4096 Jan 11 23:31 datadrwxr-xr-x 2 root root& & 4096 Jan 11 23:31 docsdrwxr-xr-x 8 root root& & 4096 Jan 11 23:31 examplesdrwxr-xr-x 6 root root& & 4096 Jan 11 23:31 lib-rw-r--r-- 1 root root&
40580 Jun& 5& 2014 LICENSE-rw-r--r-- 1 root root& & 3334 Jun& 5& 2014 NOTICE-rw-r--r-- 1 root root& & 2610 Jun& 5& 2014 README.txtdrwxr-xr-x 7 root root& & 4096 Jan 11 23:31 webappsdrwxr-xr-x 3 root root& & 4096 Jan 11 23:31 webapps-demo[root@linuxidc activemq-5.10.0]# cd bin/[root@linuxidc bin]# lltotal 152-rwxr-xr-x 1 root root 22126 Jun& 5& 2014 activemq-rwxr-xr-x 1 root root& 5665 Jun& 5& 2014 activemq-admin-rw-r--r-- 1 root root 15954 Jun& 5& 2014 activemq.jar-rwxr-xr-x 1 root root& 6189 Jun& 5& 2014 diagdrwxr-xr-x 2 root root& 4096 Jan 11 23:31 linux-x86-32drwxr-xr-x 2 root root& 4096 Jan 11 23:31 linux-x86-64drwxr-xr-x 2 root root& 4096 Jan 11 23:31 macosx-rwxr-xr-x 1 root root 83820 Jun& 5& 2014 wrapper.jar
4.启动服务
[root@linuxidc bin]# ./activemq startINFO: Using default configuration(you can configure options in one of these file: /etc/default/activemq /root/.activemqrc)
INFO: Invoke the following command to create a configuration file./activemq setup [ /etc/default/activemq | /root/.activemqrc ]
INFO: Using java '/softs/jdk1.6.0_30/bin/java'INFO: Starting - inspect logfiles specified in logging.properties and log4j.properties to get detailsINFO: pidfile created : '/softs/activemq-5.10.0/data/activemq-linuxidc.pid' (pid '28962')
5.查看是否启动成功
[root@linuxidc bin]# [root@linuxidc bin]# ps -ef | grep activemqroot& &
1 32 23:32 pts/0& & 00:00:04 /softs/jdk1.6.0_30/bin/java -Xms1G -Xmx1G -Djava.util.logging.config.file=logging.properties -Djava.security.auth.login.config=/softs/activemq-5.10.0/conf/login.config -Dcom.sun.management.jmxremote -Djava.awt.headless=true -Djava.io.tmpdir=/softs/activemq-5.10.0/tmp -Dactivemq.classpath=/softs/activemq-5.10.0/ -Dactivemq.home=/softs/activemq-5.10.0 -Dactivemq.base=/softs/activemq-5.10.0 -Dactivemq.conf=/softs/activemq-5.10.0/conf -Dactivemq.data=/softs/activemq-5.10.0/data -jar /softs/activemq-5.10.0/bin/activemq.jar startroot& &
& 0 23:32 pts/0& & 00:00:00 grep activemq[root@linuxidc bin]# [root@linuxidc bin]#
6.停止服务
[root@linuxidc data]# [root@linuxidc data]# kill 28962[root@linuxidc data]# [root@linuxidc data]# ps -ef | grep activemqroot& &
& 0 23:42 pts/0& & 00:00:00 grep activemq[root@linuxidc data]#
到此环境准备成功
package com.wzh.
import java.io.S
public class User implements Serializable{
&private static final long serialVersionUID = 1L;
&private S&&private S&&public User(String username,String password){& this.username =& this.password =&}
&public String getUsername() {&&}
&public void setUsername(String username) {& this.username =&}
&public String getPassword() {&&}
&public void setPassword(String password) {& this.password =&}
&@Override&public String toString() {& // TODO Auto-generated method stub& return "[username="+username+",password="+password+"]" ;&}
点对点： 生产者：
package com.wzh.
import java.io.S
import javax.jms.Cimport javax.jms.ConnectionFimport javax.jms.DeliveryMimport javax.jms.Dimport javax.jms.JMSEimport javax.jms.MapMimport javax.jms.Mimport javax.jms.MessagePimport javax.jms.S
import org.apache.activemq.ActiveMQCimport org.apache.activemq.ActiveMQConnectionF
public class P2PMessageProducer {
&protected String username = ActiveMQConnection.DEFAULT_USER;&protected String password = ActiveMQConnection.DEFAULT_PASSWORD;&//protected String brokerURL = "tcp://127.0.0.1:61616";&protected String brokerURL = "tcp://120.24.85.167:61616";
&protected static transient ConnectionF&protected transient C
&public static void main(String[] args) {&& try {& &new P2PMessageProducer().sendObjectMessage(new User("wzh","q123456"));& &new P2PMessageProducer().sendMapMessage();& &new P2PMessageProducer().sendTextMessage("海，你好");& } catch (Exception e) {& &e.printStackTrace();& }&}
&public P2PMessageProducer() {
& try {& &factory = new ActiveMQConnectionFactory(username, password,& & &brokerURL);& &connection = factory.createConnection();& &connection.start();& } catch (JMSException jmse) {& &close();& }&}
&/**& * 初始化连接信息& */&public P2PMessageProducer(String username, String password, String brokerURL)& &throws JMSException {& this.username =& this.password =& this.brokerURL = brokerURL;
& factory = new ActiveMQConnectionFactory(username, password, brokerURL);& connection = factory.createConnection();& try {& &connection.start();& } catch (JMSException jmse) {& &connection.close();& && }&}
&/**& * 关闭连接& */&public void close() {& try {& &if (connection != null) {& & connection.close();& &}& } catch (JMSException e) {& &e.printStackTrace();& }&}
&&protected void sendObjectMessage(Serializable serializable) throws JMSException {& Session session =& try {
& &session = connection.createSession(Boolean.TRUE,& & &Session.AUTO_ACKNOWLEDGE);& &Destination destination = session.createQueue("MessageQueue");& &MessageProducer producer = session.createProducer(destination);& &producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
& &Message message = session.createObjectMessage(serializable);
& &producer.send(message);
& } catch (JMSException e) {& &try {& & session.rollback() ;& &} catch (JMSException e1) {& & e1.printStackTrace();& &}& && } finally {& &close();& }
&protected void sendTextMessage(String text) throws JMSException {& Session session =& try {
& &session = connection.createSession(Boolean.TRUE,& & &Session.AUTO_ACKNOWLEDGE);& &Destination destination = session.createQueue("MessageQueue");& &MessageProducer producer = session.createProducer(destination);& &producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
& &Message message = session.createTextMessage(text);
& &producer.send(message);& &mit();
& } catch (JMSException e) {& &try {& & session.rollback() ;& &} catch (JMSException e1) {& & e1.printStackTrace();& &}& && } finally {& &close();& }
&}& &protected void sendMapMessage() throws JMSException {& Session session =& try {
& &session = connection.createSession(Boolean.TRUE,& & &Session.AUTO_ACKNOWLEDGE);& &Destination destination = session.createQueue("MessageQueue");& &MessageProducer producer = session.createProducer(destination);& &producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
& &MapMessage message = session.createMapMessage();& &message.setString("stock", "string");& &message.setDouble("price", 11.14);& &producer.send(message);
& } catch (JMSException e) {& &try {& & session.rollback() ;& &} catch (JMSException e1) {& & e1.printStackTrace();& &}& && } finally {& &close();& }
package com.wzh.
import javax.jms.Cimport javax.jms.ConnectionFimport javax.jms.Dimport javax.jms.JMSEimport javax.jms.MapMimport javax.jms.Mimport javax.jms.MessageCimport javax.jms.ObjectMimport javax.jms.Simport javax.jms.TextM
import org.apache.activemq.ActiveMQCimport org.apache.activemq.ActiveMQConnectionF
public class P2PMessageConsumer {
&protected String username = ActiveMQConnection.DEFAULT_USER;&protected String password = ActiveMQConnection.DEFAULT_PASSWORD;&//protected String brokerURL = "tcp://127.0.0.1:61616";&protected String brokerURL = "tcp://120.24.85.167:61616";
&protected static transient ConnectionF&protected transient C
&public static void main(String[] args) {& P2PMessageConsumer consumer = new P2PMessageConsumer();& consumer.receiveMessage();&}
&public P2PMessageConsumer() {
& try {& &factory = new ActiveMQConnectionFactory(username, password,& & &brokerURL);& &connection = factory.createConnection();& &connection.start();& } catch (JMSException jmse) {& &close();& }&}
&public P2PMessageConsumer(String username, String password, String brokerURL)& &throws JMSException {& this.username =& this.password =& this.brokerURL = brokerURL;
& factory = new ActiveMQConnectionFactory(username, password, brokerURL);& connection = factory.createConnection();& try {& &connection.start();& } catch (JMSException jmse) {& &connection.close();& && }&}
&public void close() {& try {& &if (connection != null) {& & connection.close();& &}& } catch (JMSException e) {& &e.printStackTrace();& }&}
&protected void receiveMessage() {& Session session =& try {
& &session = connection.createSession(Boolean.FALSE,& & &Session.AUTO_ACKNOWLEDGE);& &Destination destination = session.createQueue("MessageQueue");& &MessageConsumer consumer = session.createConsumer(destination);
& &while (true) {& & Message message = consumer.receive();
& & if (null != message) {
& & &if (message instanceof ObjectMessage) {& & & System.out.println("deal ObjectMessage....");& & & dealObjectMessage((ObjectMessage) message);& & &} else if (message instanceof MapMessage) {& & & System.out.println("deal MapMessage....");& & & dealMapMessage((MapMessage) message);& & &} else if (message instanceof TextMessage) {& & & System.out.println("deal TextMessage....");& & & dealTextMessage((TextMessage) message);& & &}
& & } else {& & && & }
& } catch (Exception e) {& &e.printStackTrace();& } finally {& &if (session != null) {& & try {& & &mit();& & } catch (JMSException e) {& & &e.printStackTrace();& & }& &}
&/**& * & * 处理 TextMessage消息& * & * @throws JMSException& */&private void dealTextMessage(TextMessage message) throws JMSException {& String text = message.getText();& System.out.println("text = " + text);
&/**& * & * 处理 MapMessage消息& * & * @throws JMSException& */&private void dealMapMessage(MapMessage message) throws JMSException {& String stack = message.getString("stock");& Double price = message.getDouble("price");& System.out.println("stock = " + stack + " , price =" + price);&}
&/**& * 处理ObjectMessage消息& */&private void dealObjectMessage(ObjectMessage message) throws JMSException {
& User user = (User) message.getObject();& System.out.println(user.toString());
发布与订阅： ----------------------- 消息发布者
package com.wzh.
import java.io.S
import javax.jms.Cimport javax.jms.ConnectionFimport javax.jms.DeliveryMimport javax.jms.JMSEimport javax.jms.MapMimport javax.jms.Mimport javax.jms.MessagePimport javax.jms.Simport javax.jms.T
import org.apache.activemq.ActiveMQCimport org.apache.activemq.ActiveMQConnectionF
public class Publish {
&protected String username = ActiveMQConnection.DEFAULT_USER;&protected String password = ActiveMQConnection.DEFAULT_PASSWORD;&//protected String brokerURL = "tcp://127.0.0.1:61616";&protected String brokerURL = "tcp://120.24.85.167:61616";
&protected static transient ConnectionF&protected transient C
&public static void main(String[] args) {& try {& &new Publish().sendObjectMessage(new User("wzh","q123456"));& &new Publish().sendMapMessage();& &new Publish().sendTextMessage("海，你好");& } catch (Exception e) {& &e.printStackTrace();& }&}
&public Publish() {
& try {& &factory = new ActiveMQConnectionFactory(username, password,& & &brokerURL);& &connection = factory.createConnection();& &connection.start();& } catch (JMSException jmse) {& &close();& }&}
&public Publish(String username, String password, String brokerURL)& &throws JMSException {& this.username =& this.password =& this.brokerURL = brokerURL;
& factory = new ActiveMQConnectionFactory(username, password, brokerURL);& connection = factory.createConnection();& try {& &connection.start();& } catch (JMSException jmse) {& &connection.close();& && }&}
&public void close() {& try {& &if (connection != null) {& & connection.close();& &}& } catch (JMSException e) {& &e.printStackTrace();& }&}
&protected void sendObjectMessage(Serializable serializable) throws JMSException {& Session session =& try {
& &session = connection.createSession(Boolean.TRUE,& & &Session.AUTO_ACKNOWLEDGE);& &Topic topic = session.createTopic("MessageTopic");& &MessageProducer producer = session.createProducer(topic);& &producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
& &Message message = session.createObjectMessage(serializable);
& &producer.send(message);
& } catch (JMSException e) {& &try {& & session.rollback() ;& &} catch (JMSException e1) {& & e1.printStackTrace();& &}& && } finally {& &close();& }
&protected void sendTextMessage(String text) throws JMSException {& Session session =& try {
& &session = connection.createSession(Boolean.TRUE,& & &Session.AUTO_ACKNOWLEDGE);& &Topic topic = session.createTopic("MessageTopic");& &MessageProducer producer = session.createProducer(topic);& &producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
& &Message message = session.createTextMessage(text);
& &producer.send(message);& &mit();
& } catch (JMSException e) {& &try {& & session.rollback() ;& &} catch (JMSException e1) {& & e1.printStackTrace();& &}& && } finally {& &close();& }
&}& &protected void sendMapMessage() throws JMSException {& Session session =& try {
& &session = connection.createSession(Boolean.TRUE,& & &Session.AUTO_ACKNOWLEDGE);& &Topic topic = session.createTopic("MessageTopic");& &MessageProducer producer = session.createProducer(topic);& &producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
& &MapMessage message = session.createMapMessage();& &message.setString("stock", "string");& &message.setDouble("price", 11.14);& &producer.send(message);
& } catch (JMSException e) {& &try {& & session.rollback() ;& &} catch (JMSException e1) {& & e1.printStackTrace();& &}& && } finally {& &close();& }
消息订阅者：
package com.wzh.
import javax.jms.Cimport javax.jms.ConnectionFimport javax.jms.Dimport javax.jms.JMSEimport javax.jms.MapMimport javax.jms.Mimport javax.jms.MessageCimport javax.jms.MessageLimport javax.jms.MessagePimport javax.jms.ObjectMimport javax.jms.Simport javax.jms.TextMimport javax.jms.T
import org.apache.activemq.ActiveMQCimport org.apache.activemq.ActiveMQConnectionF
public class Subscriber {
&protected String username = ActiveMQConnection.DEFAULT_USER;&protected String password = ActiveMQConnection.DEFAULT_PASSWORD;&//protected String brokerURL = "tcp://127.0.0.1:61616";&protected String brokerURL = "tcp://120.24.85.167:61616";
&protected static transient ConnectionF&protected transient C
&public static void main(String[] args) {& Subscriber consumer = new Subscriber();& consumer.receiveMessage();&}
&public Subscriber() {
& try {& &factory = new ActiveMQConnectionFactory(username, password,& & &brokerURL);& &connection = factory.createConnection();& &connection.start();& } catch (JMSException jmse) {& &close();& }&}
&public Subscriber(String username, String password, String brokerURL)& &throws JMSException {& this.username =& this.password =& this.brokerURL = brokerURL;
& factory = new ActiveMQConnectionFactory(username, password, brokerURL);& connection = factory.createConnection();& try {& &connection.start();& } catch (JMSException jmse) {& &connection.close();& && }&}
&public void close() {& try {& &if (connection != null) {& & connection.close();& &}& } catch (JMSException e) {& &e.printStackTrace();& }&}
&protected void receiveMessage() {& Session session =& try {
& &session = connection.createSession(Boolean.FALSE,& & &Session.AUTO_ACKNOWLEDGE);& &Topic topic = session.createTopic("MessageTopic");& &MessageConsumer consumer = session.createConsumer(topic);& && &consumer.setMessageListener(new MessageListener() {& & & & @Override& & public void onMessage(Message message) {
& & &if (message instanceof ObjectMessage) {& & & System.out.println("deal ObjectMessage....");& & & dealObjectMessage((ObjectMessage) message);& & &} else if (message instanceof MapMessage) {& & & System.out.println("deal MapMessage....");& & & dealMapMessage((MapMessage) message);& & &} else if (message instanceof TextMessage) {& & & System.out.println("deal TextMessage....");& & & dealTextMessage((TextMessage) message);& & &}& & && & }& &}) ;
& } catch (Exception e) {& &e.printStackTrace();& } finally {& &/*if (session != null) {& & try {& & &mit();& & } catch (JMSException e) {& & &e.printStackTrace();& & }& &}*/
&/**& * & * 处理 TextMessage消息& * & * @throws JMSException& */&private void dealTextMessage(TextMessage message) {& try {& &String text = message.getText();& &System.out.println("text = " + text);& } catch (JMSException e) {& &e.printStackTrace();& }
&/**& * & * 处理 MapMessage消息& * & * @throws JMSException& */&private void dealMapMessage(MapMessage message){& try {& &String stack = message.getString("stock");& &Double price = message.getDouble("price");& &System.out.println("stock = " + stack + " , price =" + price);& } catch (JMSException e) {& &e.printStackTrace();& }&}
&/**& * 处理ObjectMessage消息& */&private void dealObjectMessage(ObjectMessage message){
& try {& &User user = (User) message.getObject();& &System.out.println(user.toString());& } catch (JMSException e) {& &e.printStackTrace();& }
推荐阅读：
Linux系统下ActiveMQ 安装
下的ACTIVEMQ服务器
6.5启动ActiveMQ报错解决
Spring+JMS+ActiveMQ+Tomcat实现消息服务
Linux环境下面ActiveMQ端口号设置和WEB端口号设置
本文永久更新链接地址：
相关资讯 & & &
& (03月13日)
& (12/26/:47)
& (12/27/:30)
& (11/12/:16)
　　　同意评论声明
　　　发表
尊重网上道德，遵守中华人民共和国的各项有关法律法规
承担一切因您的行为而直接或间接导致的民事或刑事法律责任
本站管理人员有权保留或删除其管辖留言中的任意内容
本站有权在网站内转载或引用您的评论
参与本评论即表明您已经阅读并接受上述条款}