特点(作用)

根据消息队列的特点可以衍生出很多场景或者说很多场景都能用到。

下面举几个例子

2应用解耦

是Apache出品最流行的能力强劲的开源消息总线。

ActiveMQ

JMS

Provider实现,尽管JMS规范出台已经是很久的事情了,但是JMS在当今的J2EE应用中间仍然扮演着特殊的地位。

支持来自JavaCC

Spring支持以便ActiveMQ可以轻松嵌入到Spring应用程序中并使用Spring的XML配置机制进行配置

专为高性能集群客户端

CXF和Axis支持以便ActiveMQ可以轻松地放入这些Web服务堆栈中以提供可靠的消息传递

支持可插拔传输协议例如in-VMTCPSSLNIOUDP多播JGroups和JXTA传输

2.RabbitMQ

RabbitMQ可以部署在分布式和联合配置中以满足高规模高可用性需求。

RabbitMQ可运行在许多操作系统和云环境中并为大多数流行语言提供广泛的开发工具。

来自官网翻译

AMQP

MessageQueue高级消息队列协议。

它是应用层协议的一个开放标准为面向消息的中间件设计基于此协议的客户端与消息中间件可传递消息并不受产品、开发语言等条件的限制。

RabbitMQ最初广泛应用于金融行业根据官网描述它具有如下特点

异步消息传递支持多种消息协议消息队列传送确认灵活的路由到队列多种交换类型

CRubyPerlPythonPHP等等

Kafka是由Apache软件基金会开发的一个开源流处理平台由Scala和Java编写。

Kafka是一种高吞吐量的分布式发布订阅消息系统它可以处理消费者规模的网站中的所有动作流数据。

这种动作网页浏览搜索和其他用户的行动是在现代网络上的许多社会功能的一个关键因素。

这些数据通常是由于吞吐量的要求而通过处理日志和日志聚合来解决。

对于像Hadoop的一样的日志数据和离线分析系统但又要求实时处理的限制这是一个可行的解决方案。

Kafka的目的是通过Hadoop的并行加载机制来统一线上和离线的消息处理也是为了通过集群来提供实时的消息。

Kafka它主要用于处理活跃的流式数据因此Kafaka在大数据系统中使用较多。

同时为发布和订阅提供高吞吐量。

据了解Kafka每秒可以生产约25万消息50

MB每秒处理55万消息110

可进行持久化操作。

将消息持久化到磁盘因此可用于批量消费例如ETL以及实时应用程序。

通过将数据持久化到硬盘以及replication防止数据丢失。

分布式系统易于向外扩展。

所有的producer、broker和consumer都会有多个均为分布式的。

无需停机即可扩展机器。

消息被处理的状态是在consumer端维护而不是由server端维护。

当失败时能自动平衡。

RocketMQ

RocketMQ是阿里开源的消息中间件目前在Apache孵化使用纯Java开发具有高吞吐量、高可用性、适合大规模分布式系统应用的特点。

RocketMQ思路起源于Kafka但并不是简单的复制它对消息的可靠传输及事务性做了优化目前在阿里集团被广泛应用于交易、充值、流计算、消息推送、日志流式处理、binglog分发等场景支撑了阿里多次双十一活动。

支持拉pull和推push两种消息模式

activemq在各个系统下都有对应的安装包。

以下来演示Linux系统下安装activemq。

进入apache-activemq-5.15.8/bin目录启动activemq./activemq

start

待ActiveMQ安装启动好访问http://ip:8161/admin登录名和密码都是admin(在配置文件中可修改)进入ActiveMQ的主页

1.Queue消息队列页面

Dequeued出了队列的消息可以理解为是被消费掉的消息数量。

在Queues里它和进入队列的总数量相等(因为一个消息只会被成功消费一次),如果暂时不等是因为消费者还没来得及消费。

2.Topic主题页面

Dequeued出了队列的消息可以理解为是被消费掉的消息数量。

在Topics里因为多消费者从而导致数量会比入队列数高。

3.Subscribers查看订阅者页面

groupIdorg.apache.activemq/groupId

artifactIdactivemq-core/artifactId

version5.7.0/version

点对点模型采用的是队列(Queue)作为消息载体。

在该模式中一条消息只能被一个消费者消费没有被消费的只能留在队列中等待被消费或者超时。

举个例子如果队列中有10条消息有两个消费者就是一个消费者消费5条信息你一条我一条。

以下以代码演示。

消息发布者

1.建立ConnectionFactory工厂对象需要填入用户名、密码、连接地址一般使用默认如果没有修改的话

2.通过ConnectionFactory对象创建一个Connection连接并且调用Connection的start方法开启连接Connection方法默认是关闭的

3.通过Connection对象创建Session会话上下文环境对象用于接收消息参数1是是否启用事物参数2是签收模式一般设置为自动签收

4.通过Session对象创建Destination对象指的是一个客户端用来制定生产消息目标和消费消息来源的对象。

在PTP的模式中Destination被称作队列在Pub/Sub模式中Destination被称作主题Topic

5.通过Session对象创建消息的发送和接收对象生产者和消费者

6.通过MessageProducer的setDeliverMode方法为其设置持久化或者非持久化特性

7.使用JMS规范的TextMessage形式创建数据通过Session对象并用MessageProducer的send方法发送数据。

客户端同理。

记得关闭

ConnectionFactory

ActiveMQConnectionFactory(ActiveMQConnectionFactory.DEFAULT_USER,

ActiveMQConnectionFactory.DEFAULT_PASSWORD,tcp://94.191.49.192:61616);

Connection

connectionFactory.createConnection();

connection.start();

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

Destination