JMS消息传送机制ActiveMQ

JMS(Java Message Service) 消息服务是一个面向消息中间件（MOM）的API，用于在两个应用程序之间，或者分布式系统中发送消息消费消息，进行异步通信，数据交互。JMS提供的API与具体厂商无关，抽象出接口提供给厂商去实现。类似于JDBC是可以用来访问许多不同关系数据库的
API.JMS 使您能够通过消息收发服务（有时称为消息中介程序或路由器）从一个 JMS 客户机向另一个 JMS客户机发送消息，并且通过提供标准的产生、发送、接收消息的接口简化企业应用的开发。

消息：

消息是一种类型的对象,由两部分组成：报文头和消息主体。报头由路由信息以及有关该消息的元数据组成。消息主体则携带着应用程序的数据。



 

对象模型：

JMS连接工厂：（ConnectionFactory）

JMS连接：(Connection)

JMS会话：(Session)

JMS目的地：(Destination)

JMS生产者：(Message Procedur)

JMS消费者：(Message Consumer)



模型：

JMS 支持两类消息传送模型（消息传送域）：点对点模型和发布/订阅模型

点对点（Point-to-Point）。在点对点的消息系统中，消息分发给一个单独的使用者。点对点消息往往与队列（javax.jms.Queue）相关联,是一对一的消息模型。

发布/订阅（Publish/Subscribe）。发布/订阅消息系统支持一个事件驱动模型，消息生产者和消费者都参与消息的传递。生产者发布事件，而使用者订阅感兴趣的事件，并使用事件。该类型消息一般与特定的主题（javax.jms.Topic）关联，是一对多的模型.



 

ActiveMQ



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

整体架构

ActiveMQ主要涉及到5个方面：



传输协议
消息之间的传递，无疑需要协议进行沟通，启动一个ActiveMQ打开了一个监听端口，
ActiveMQ提供了广泛的连接模式，其中主要包括SSL、STOMP、XMPP；ActiveMQ默认的使用的协议是openWire，端口号：61616;

消息域
ActiveMQ主要包含Point-to-Point (点对点),Publish/Subscribe Model (发布/订阅者)，其中在Publich/Subscribe 模式下又有Nondurable subscription和     durable
subscription (持久化订阅)2种消息处理方式

消息存储
在消息传递过程中，部分重要的消息可能需要存储到数据库或文件系统中，当中介崩溃时，信息不回丢失

Cluster  (集群)
最常见到 集群方式包括network of brokers和Master Slave；

Monitor (监控)
ActiveMQ一般由jmx来进行监控；

安装运行

    
下载地址：http://activemq.apache.org/download.html 下载：apache-activemq-5.13.3-bin.tar.gz,解压

  




  

   检验安装成果：http://localhost:8161/admin/  用户名：admin 密码 admin

 


 

  

  ActiveMQ HelloWord:

 生产者：

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

public class Send
{
public static final String QueueName = "MessageQueue";

public static void main(String[] args) throws JMSException
{
//1. 创建ConnctionFactroy 工厂对象，需要用户名 密码 链接的地址
ConnectionFactory facotry = new ActiveMQConnectionFactory(ActiveMQConnectionFactory.DEFAULT_USER, ActiveMQConnectionFactory.DEFAULT_PASSWORD, "tcp://localhost:61616");
//2. 通过ConnectionFactory工厂对象创建Connection,并且调用start()方法开启连接，默认是关闭状态
Connection connection = facotry.createConnection();
connection.start();
//3. 通过Connection创建会话，用于发送、接收消息。args0(是否开启事务) args1(签收的模式)
Session session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
//4 .创建Destination 对象, 生产消息目标或消费消息来源的的对象，在PTP中是Queue对象，在Pusb/Sec 是Topic对象
Destination destination = session.createQueue(QueueName);
//⑤ 通过Session创建生产者或者消费者
MessageProducer producer = session.createProducer(destination);
//⑥ 设置消息的持久化或非持久化 持久化方式（KaHaDB JDBC[MYSQL ORACLE] ）
producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
//⑦ 创建消息
TextMessage message = session.createTextMessage();
message.setText("hello ActiveMQ");
int i = 0;
while (i < 7)
{
producer.send(message);
i++;
}

if (connection != null)
{
connection.close();
}
}
}

消费者：

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.MessageConsumer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

public class Reveice
{
public static void main(String[] args) throws Exception
{
ConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");
Connection connection = factory.createConnection(ActiveMQConnectionFactory.DEFAULT_USER, ActiveMQConnectionFactory.DEFAULT_PASSWORD);
connection.start();
Session session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);

Destination destination = session.createQueue(Send.QueueName);

MessageConsumer consumer = session.createConsumer(destination);

while (true)
{
TextMessage message = (TextMessage) consumer.receive();
System.out.println(message.getText());
}
}
}

 



 



 

本章完结