activemq后台管理 看topic消息_ActiveMQ学习笔记

ActiveMQ

1.下载安装

目录结构：

bin存放的是脚本文件

conf存放的是基本配置文件

data存放的是日志文件

docs存放的是说明文档

examples存放的是简单的实例

lib存放的是activemq所需jar包

webapps用于存放项目的目录

2.启动

进入到ActiveMQ 安装目录的Bin 目录，linux 下输入 ./activemq start 启动activeMQ 服务。

输入命令之后，会提示我们创建了一个进程IP 号，这时候说明服务已经成功启动了。

ActiveMQ默认启动时，启动了内置的jetty服务器，提供一个用于监控ActiveMQ的admin应用。

admin：

http://127.0.0.1:8161/admin/​127.0.0.1:8161

我们在浏览器打开链接之后输入账号密码

默认账号：admin

密码：admin

ActiveMQ 在linux 下的终止命令是 ./activemq stop

发送消息的基本步骤：

  • 1.创建连接使用的工厂类JMS ConnectionFactory
  • 2.使用管理对象JMS ConnectionFactory建立连接Connection,并启动
  • 3.使用连接Connection建立会话Session
  • 4.使用会话Session和管理对象Destination创建消息生产者MessageSendr
  • 5.使用消息生产者MessageSender发送消息

消息接收者从JMS接收消息的步骤：

  • 1.创建连接使用的工厂类JMS ConnectionFactory
  • 2.使用管理对象JMS ConnectionFactory建立连接Connection，并启动
  • 3.使用连接Connection 建立会话Session
  • 4.使用会话Session和管理对象Destination创建消息接收者MessageReceive
  • 5.使用消息接收者MessageReceiver接受消息，需要用setMessageListener将MessageListener接口绑定到MessageReceiver消息接收者必须实现了MessageListener接口，需要定义onMessage事件方法。

JMQ的两种消息模式

1.点对点模式PTP

点对点的模式主要建立在一个队列上面，当连接一个列队的时候，发送端不需要知道接收端是否正在接收，可以直接向ActiveMQ发送消息，发送的消息，将会先进入队列中，如果有接收端在监听，则会发向接收端，如果没有接收端接收，则会保存在activemq服务器，直到接收端接收消息，点对点的消息模式可以有多个发送端，多个接收端，但是一条消息，只会被一个接收端给接收到，哪个接收端先连上ActiveMQ，则会先接收到，而后来的接收端则接收不到那条消息

2.订发布阅模式Pub/Sub

订阅/发布模式，同样可以有着多个发送端与多个接收端，但是接收端与发送端存在时间上的依赖，就是如果发送端发送消息的时候，接收端并没有监听消息，那么ActiveMQ将不会保存消息，将会认为消息已经发送，换一种说法，就是发送端发送消息的时候，接收端不在线，是接收不到消息的，哪怕以后监听消息，同样也是接收不到的。这个模式还有一个特点，那就是，发送端发送的消息，将会被所有的接收端给接收到，不类似点对点，一条消息只会被一个接收端给接收到。

JMS JAVA Message Service java消息服务

JMS的级别组件

  • 1.连接工厂
  • 2.连接
  • 3.会话
  • 4.目的地
  • 5.消息生产者
  • 6.消息消费者
  • 7.消息

JMS的可靠性机制

  • 1.确认
  • 2.持久性
  • 3.优先级
  • 4.消息过期
  • 5.临时目的地
  • 6.持久订阅
  • 7.本地事务

Sendre/Receiver:

  • 1.建立ConnectionFactory工厂对象，需要填入用户名、密码、以及要连接的地址，均使用默认的即可，默认端口为tcp://localhost:61616
  • 2.通过ConnectionFactory工厂对象创建一个Connection连接，并且调用Connection的start方法开启连接，Connection默认是关闭的
  • 3.通过Connection对象创建Session会话(上下文环境对象)，用于接收消息，参数配置1为是否启用事务 参数配置2为签收模式，一般我们设置自动签收
  • 4.通过Session创建Destination对象,指的是一个客户端用了指定生产消息目标和消费消息来源对象，PTP模式中，Destination被称为Queue队列；在Pub/Sub模式，被称为Topic即主题，在程序中可以使用多个Queue和Topic
  • 5.我们需要通过Session对象创建消息的发送和接收对象(生产者和消费者)MessageProducer/MessageConsumer
  • 6.我们可以使用MessageProducer的setDeliverMode方法为其设置持久化特性和非持久化特性(DeliverMode).
  • 7.最后我们使用JMS规范的TextMessage形式创建数据(通过Session对象)，并用MessageProducer的send方法发送数据。同理客户端使用receive方法进行接收数据，最后不要忘记关闭Connection连接

ActiveMQ的Web管理界面http://127.0.0.1:8161/admin

activemq管控台使用jetty部署，所以需要修改密码则需要到相应的配置文件中

activemq应该设有安全机制，只有符合认证的用户才能进行发送和获取消息，所以可以在activemq.xml里添加安全验证配置。在123行之后添加

JMS术语

  • 1.Producer:生产者
  • 2.Consumer:消费者
  • 3.PTP:点对点的消息模式
  • 4.Pub/Sub:发布/订阅的消息类型
  • 5.Queue:队列目标
  • 6.Topic:主题目标
  • 7.ConnectionFactory:连接工厂 JMS用他创建连接对象 以连接到JMS的provider
  • 8.Connection:JMS客户端到JMS Producer的连接
  • 9.Destination:消息的目的地
  • 10.Session:会话 一个发送或接收消息的线程 会话用于创建消息生产者、消息消费者和消息等，会话提供了一个事务性的上下文，在这个上下文中，一组发送和接收被组合都一个原子操作中
  • 11.Acknowledge：签收
  • 12.Transaction：事务

JMS Message：JMS的消息 由以下三部分组成

  • 1.消息头：每个消息头字段都有相应的getter和 setter方法
  • 2.消息属性：如果需要除消息头字段以外的值，那么就使用消息属性
  • 3.消息体：封装具体的消息数据

JMS定义了5种不同类型的消息正文格式（消息体）

  • 1.StreamMessage java原始值的数据流
  • 2.MapMessage 一套名称-值对
  • 3.TextMessage 一个字符串对象
  • 4.ObjectMessage 一个序列化的java对象
  • 5.ByteMessage 一个未解释字节的数据流

JMS Producer:消息生产者 创建和发送JMS消息的客户端应用

JMS Consumer:消息消费者 接收和处理JMS消息的客户端应用

消息的消费可以采用以下两种方法之一：

  • 1.同步消费：通过调用消费者的receive方法从目的地中显示提取消息，receive方法可以一直阻塞到消息到达
  • 2.异步消费：客户可以为消费者注册一个消息监听器，以定义在消息到达时所采取的动作：

消息存储持久化

不仅支持persistent和non-persistent两种方式，还支持消息的恢复(recovery)方式

PTP Queue的存储是很简单的 就是一个FIFO的Queue

pub/sub :

对于持久化订阅主题，每一个消费者将获得一个消息的复制，

有效的消息存储

ActiveMQ提供了一个插件式的消息存储，类似于消息的多点传播，主要实现了如下几种：

  • 1.AMQ消息存储 基于文件的存储方式，是以前的默认的消息存储
  • 2.KahaDB消息存储 提供了容量的提升和恢复能力，是现在的默认存储方式
  • 3.JDBC消息存储 消息基于JDBC存储
  • 4.Memory消息存储 基于内存的消息存储

JMS可靠性机制

4262d4645f5579284b5d98a89a29f90d.png

800f68fdc9dafd953ed1fbedac8c6b57.png

76e989c20c090275aee44cb4b822147f.png

ac695e32306b8cc68832374a2734d044.png

JMS的API结构和开发步骤

f38f593a00a76349cf57fb93a772fda4.png

418a37173c61c1873055d9b3d2afe4b5.png

ActiveMQ结合spring

18e19816720cb03d82acbd6188e56b93.png

718b23d65b1c8036652805cbd50f0dad.png

queue消息发送

2ce75da89e5e5670adacf76c2d894704.png

queue消息接收

d409e9affd04a4e8c1ac3c7f43714340.png

如果topic的话

ActiveMQ支持的传输协议

Transmission Control Protocol TCP

1.这是默认的broker的配置，TCP监听端口是61616

2.在网络传输数据前，必须要序列化数据，消息通过一个叫write protocol来序列化成字节流，默认情况下，ActiveMQ把write protocol叫做OpenWire,她的目的是促使网络上的效率和数据快速交互

3.TCP连接的URI形式：tcp://hostname:port?key=value&key=value ?之前是必须的

4.TCP传输特点：

a.TCP协议传输可靠性高，稳定性强

b.高效率：字节流方式传递，效率很高

c.有效性、可用性：应用广泛

5.所有关于Transport协议的可配置参数，可参考官方文档

New I/O API Protocol (NIO)

1.nio协议和tcp协议类似，但nio更侧重底层的访问操作。他允许开发人员对同一资源可有更多的client调用和服务器有更多的负载。

2.适合使用nio的场景

a.可能有大量的Client去连接到broker上 一般情况下 大量的client去链接broker是被操作系统的线程数限制的。因此NIO的实现比TCP需要更少的线程去执行

b.可能对于broker有一个迟钝的网络传输 NIO比TCP提供更好的性能

3.NIO连接的URI形式：nio://hostname:port?key=value

4.Transport Connector配置

<transportConnectors>

用ActiveMQ构建应用

Borker:相当于一个ActiveMQ服务器实例

命令行启动参数示例如下：

  • 1.activemq start:使用默认的activemq.xml来启动
  • 2.activemq start xbean:file:../conf/activemq-2.xml 使用指定的配置文件
  • 3.如果不指定file,也就是xbean:activemq-2.xml,那么xml必须在classpath下面

用ActiveMQ构建Java应用

这里主要将用ActiveMQ Borker 作为独立的消息服务器来构建JAVA应用。ActiveMQ也支持在vm中通信基于嵌入式的broker,能够无缝集成其他java应用

嵌入式Broker启动

1：Broker Service启动broker，示例如下：

BrokerService

2：BrokerFactory启动broker:示例如下：

String

3：broker.properties的内容如下

useJmx=true
persistent=false
borkerName=Cheese

利用Spring集成Broker,Spring的配置文件如下

<beans>

ActiveMQ的静态网络连接

ActiveMQ的networkConnector

在某些场景下，需要多个ActiveMQ的Broker做集群，那么就涉及到了Broker到Broker的通信，这个被称为ActiveMQ的networkConnector

ActiveMQ的networkConnector默认是单向的，一个Broker在一端发送消息，另一个Broker在另一端接收消息，这就是所谓的桥接。ActiveMQ也支持双向链接，创建一个双向的通道对于两个Broker,不仅发送消息也能从相同的通道来接收消息，通常作为duplex connector来映射，如图：

7f9e87beb7908d89c79d4a6f33632067.png

discovery的概念

一般情况下，discovery是被用来发现远程的服务，客户端通常想去发现所有可利用的brokers;另一层意思，它是基于现有的网络Broker去发现其他可用的Brokers.

有两种配置Client到Broker的链接方式，一种方式：Client通过Statically配置的方式去连接Broker;第二种方式：Client通过discovery agents来dynamically的发现Brokers

Static networks

Static networkConnector是用于创建一个静态的配置对于网络中的多个Broker.这种协议用于复合url，一个复合url包括多个url地址。格式如下：

static:(uri1,uri2,uri3,...)?key=value

1.配置如下

<networkConnectors>