消息中间件解决方案ActiveMQ

ゞ浴缸里的玫瑰 2023-02-14 07:43 12阅读 0赞

# ActiveMQ入门 #

## 1.1 JMS简介 ##

JMS（Java Messaging Service）是Java平台上有关面向消息中间件的技术规范，它便于消息系统中的Java应用程序进行消息交换,并且通过提供标准的产生、发送、接收消息的接口简化企业应用的开发。  
JMS本身只定义了一系列的接口规范，是一种与厂商无关的 API，用来访问消息收发系统。它类似于 JDBC(java Database Connectivity)：这里，JDBC 是可以用来访问许多不同关系数据库的 API，而 JMS 则提供同样与厂商无关的访问方法，以访问消息收发服务。许多 厂商目前都支持 JMS，包括 IBM 的 MQSeries、BEA的 Weblogic JMS service和 Progress 的 SonicMQ，这只是几个例子。 JMS 使您能够通过消息收发服务（有时称为消息中介程序或路由器）从一个 JMS 客户机向另一个 JML 客户机发送消息。消息是 JMS 中的一种类型对象，由两部分组成：报头和消息主体。报头由路由信息以及有关该消息的元数据组成。消息主体则携带着应用程序的数据或有效负载。根据有效负载 的类型来划分，可以将消息分为几种类型，它们分别携带：简单文本 (TextMessage)、可序列化的对象 (ObjectMessage)、属性集合 (MapMessage)、字节流 (BytesMessage)、原始值流 (StreamMessage)，还有无有效负载的消息 (Message)。

## 1.2ActiveMQ简介 ##

ActiveMQ 是Apache出品，最流行的，能力强劲的开源消息总线。ActiveMQ 是一个完全支持JMS1.1和J2EE 1.4规范的 JMS Provider实现，尽管JMS规范出台已经是很久的事情了，但是JMS在当今的J2EE应用中间仍然扮演着特殊的地位。  
对于消息的传递有两种类型：  
一种是点对点的，即一个生产者和一个消费者一一对应；  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Noa3N0YXJ0_size_16_color_FFFFFF_t_70]  
另一种是发布/ 订阅模式，即一个生产者产生消息并进行发送后，可以由多个消费者进  
行接收。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Noa3N0YXJ0_size_16_color_FFFFFF_t_70 1]  
JMS 定义了五种不同的消息正文格式，以及调用的消息类型，允许你发送并接收以一  
些不同形式的数据，提供现有消息格式的一些级别的兼容性。  
· StreamMessage – Java 原始值的数据流  
· MapMessage–一套名称-值对  
· TextMessage–一个字符串对象  
· ObjectMessage–一个序列化的 Java 对象  
· BytesMessage–一个字节的数据流

## 1.4ActiveMQ下载与安装 ##

### 1.4.1下载 ###

官方网站下载：http://activemq.apache.org/

### 1.4.2安装（Linux） ###

（1）将apache-activemq-5.12.0-bin.tar.gz 上传至服务器  
（2）解压此文件

tar  zxvf  apache-activemq-5.12.0-bin.tar.gz

（3）为apache-activemq-5.12.0目录赋权

chmod 777 apache-activemq-5.12.0

（4）进入apache-activemq-5.12.0\\bin目录  
（5）赋与执行权限

chmod 755 activemq

\-------------------------------------- 知识点小贴士 --------------------------  
linux 命令chmod 755的意思  
chmod是Linux下设置文件权限的命令，后面的数字表示不同用户或用户组的权限。  
一般是三个数字：  
第一个数字表示文件所有者的权限  
第二个数字表示与文件所有者同属一个用户组的其他用户的权限  
第三个数字表示其它用户组的权限。  
权限分为三种：读（r=4），写（w=2），执行（x=1） 。 综合起来还有可读可执行（rx=5=4+1）、可读可写（rw=6=4+2）、可读可写可执行(rwx=7=4+2+1)。  
所以，chmod 755 设置用户的权限为：  
1.文件所有者可读可写可执行 --7  
2.与文件所有者同属一个用户组的其他用户可读可执行 --5  
3.其它用户组可读可执行 --5

### 1.4.3启动 ###

./activemq start

出现下列提示表示成功！  
![在这里插入图片描述][20200601173340844.png]  
假设服务器地址为192.168.25.129 ，打开浏览器输入地址  
http://192.168.25.129:8161/ 即可进入ActiveMQ管理页面  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Noa3N0YXJ0_size_16_color_FFFFFF_t_70 2]  
点击进入管理页面  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Noa3N0YXJ0_size_16_color_FFFFFF_t_70 3]  
输入用户名和密码 均为 admin  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Noa3N0YXJ0_size_16_color_FFFFFF_t_70 4]  
进入主界面  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Noa3N0YXJ0_size_16_color_FFFFFF_t_70 5]

## 1.5 ActiveMQ入门小DEMO（点对点） ##

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

### 1.5.1消息生产者 ###

（1）创建工程activemqDemo ，引入依赖

<dependency>
    	  		<groupId>org.apache.activemq</groupId>
    	  		<artifactId>activemq-all</artifactId>
    	  		<version>5.11.2</version>
    	  	</dependency>

（2）创建类QueueProducer

package cn.atguigu.demo;
    import javax.jms.Connection;
    import javax.jms.JMSException;
    import javax.jms.MessageProducer;
    import javax.jms.Queue;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    import org.apache.activemq.ActiveMQConnectionFactory;
    /** * 点对点（消息生产者） * @author Administrator * */
    public class QueueProducer { 
    	public static void main(String[] args) throws JMSException { 
    		//1.创建连接工厂
    		ActiveMQConnectionFactory connectionFactory=new ActiveMQConnectionFactory("tcp://192.168.25.129:61616");
    		//2.获取连接
    		Connection connection = connectionFactory.createConnection();
    		//3.启动连接
    		connection.start();
    		//4.获取session
    		Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    		//5.创建队列对象
    		Queue queue = session.createQueue("test-queue");
    		//6.创建消息生产者对象
    		MessageProducer producer = session.createProducer(queue);
    		//7.创建消息
    		TextMessage textMessage = session.createTextMessage("欢迎来到神奇的品优购世界！");
    		//8.使用生成者发送消息
    		producer.send(textMessage);
    		//9.关闭资源
    		producer.close();
    		session.close();
    		connection.close();
    	}
    }

上述代码中第4步创建session 的两个参数：  
第1个参数 是否使用事务  
第2个参数 消息的确认模式  
AUTO\_ACKNOWLEDGE = 1 自动确认  
CLIENT\_ACKNOWLEDGE = 2 客户端手动确认  
DUPS\_OK\_ACKNOWLEDGE = 3 自动批量确认  
SESSION\_TRANSACTED = 0 事务提交并确认  
运行后通过ActiveMQ管理界面查询  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Noa3N0YXJ0_size_16_color_FFFFFF_t_70 6]

### 1.5.2消息消费者 ###

创建类

package cn.atguigu.demo;
    import java.io.IOException;
    import javax.jms.Connection;
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageConsumer;
    import javax.jms.MessageListener;
    import javax.jms.Queue;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    import org.apache.activemq.ActiveMQConnectionFactory;
    /** * 消息消费者 * @author Administrator * */
    public class QueueConsumer { 
    	public static void main(String[] args) throws JMSException, IOException { 
    		//1.创建连接工厂
    		ActiveMQConnectionFactory connectionFactory=new ActiveMQConnectionFactory("tcp://192.168.25.129:61616");		
    		//2.获取连接
    		Connection connection = connectionFactory.createConnection();
    		//3.启动连接
    		connection.start();
    		//4.获取session
    		Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    		//5.创建队列对象
    		Queue queue = session.createQueue("test-queue");
    		//6.创建消息消费者
    		MessageConsumer consumer=session.createConsumer(queue);
    		//7.接受消息
    		consumer.setMessageListener(new MessageListener() { 			
    			@Override
    			public void onMessage(Message message) { 
    				//获取文本消息对象
    				TextMessage textMessage=(TextMessage)message;				
    				try { 
    					String text = textMessage.getText();//提取文本
    					System.out.println(text);
    				} catch (JMSException e) { 
    					// TODO Auto-generated catch block
    					e.printStackTrace();
    				}				
    			}
    		});
    		//8.等待键盘输入
    		System.in.read();
    		//9.关闭资源 
    		consumer.close();
    		session.close();
    		connection.close();
    	}
    }

执行后看到控制台输出

## 1.6 ActiveMQ入门小DEMO（发布/订阅） ##

### 1.6.1消息生产者 ###

创建类

package cn.atguigu.demo;
    import javax.jms.Connection;
    import javax.jms.JMSException;
    import javax.jms.MessageProducer;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    import javax.jms.Topic;
    import org.apache.activemq.ActiveMQConnectionFactory;
    /** * 消息订阅模式-生产者 * @author Administrator * */
    public class TopicProducer { 
    	public static void main(String[] args) throws JMSException { 		
    		//1.创建连接工厂
    		ActiveMQConnectionFactory connectionFactory=new ActiveMQConnectionFactory("tcp://192.168.25.129:61616");		
    		//2.创建连接
    		Connection connection = connectionFactory.createConnection();		
    		//3.启动连接
    		connection.start();
    		//4.创建会话
    		Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    		//5.创建一个订阅
    		Topic topic = session.createTopic("test-topic");
    		//6.创建消息生产者 
    		MessageProducer producer = session.createProducer(topic);
    		//7.创建消息
    		TextMessage textMessage = session.createTextMessage("您的手机该续费了！");
    		//8.发送消息 
    		producer.send(textMessage);
    		//9.关闭资源
    		producer.close();
    		session.close();
    		connection.close();
    	}
    }

运行效果如下：  
![在这里插入图片描述][20200601173648879.png]

### 1.6.2消息消费者 ###

代码：

package cn.atguigu.demo;
    import java.io.IOException;
    import javax.jms.Connection;
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageConsumer;
    import javax.jms.MessageListener;
    import javax.jms.Queue;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    import javax.jms.Topic;
    import org.apache.activemq.ActiveMQConnectionFactory;
    public class TopicConsumer { 
    	public static void main(String[] args) throws JMSException, IOException { 
    		//1.创建连接工厂
    		ActiveMQConnectionFactory connectionFactory=new ActiveMQConnectionFactory("tcp://192.168.25.129:61616");		
    		//2.获取连接
    		Connection connection = connectionFactory.createConnection();
    		//3.启动连接
    		connection.start();
    		//4.获取session
    		Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    		//5.创建主题对象
    		Topic topic = session.createTopic("test-topic");
    		//6.创建消息消费者
    		MessageConsumer consumer=session.createConsumer(topic);
    		//7.接受消息
    		consumer.setMessageListener(new MessageListener() { 			
    			@Override
    			public void onMessage(Message message) { 
    				//获取文本消息对象
    				TextMessage textMessage=(TextMessage)message;				
    				try { 
    					String text = textMessage.getText();//提取文本
    					System.out.println(text);
    				} catch (JMSException e) { 
    					// TODO Auto-generated catch block
    					e.printStackTrace();
    				}				
    			}
    		});
    		//8.等待键盘输入
    		System.in.read();
    		//9.关闭资源 
    		consumer.close();
    		session.close();
    		connection.close();
    	}
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Noa3N0YXJ0_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200601173110368.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Noa3N0YXJ0,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Noa3N0YXJ0_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200601173118833.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Noa3N0YXJ0,size_16,color_FFFFFF,t_70
[20200601173340844.png]: https://img-blog.csdnimg.cn/20200601173340844.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Noa3N0YXJ0_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20200601173349667.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Noa3N0YXJ0,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Noa3N0YXJ0_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20200601173407940.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Noa3N0YXJ0,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Noa3N0YXJ0_size_16_color_FFFFFF_t_70 4]: https://img-blog.csdnimg.cn/20200601173417542.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Noa3N0YXJ0,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Noa3N0YXJ0_size_16_color_FFFFFF_t_70 5]: https://img-blog.csdnimg.cn/20200601173431151.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Noa3N0YXJ0,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Noa3N0YXJ0_size_16_color_FFFFFF_t_70 6]: https://img-blog.csdnimg.cn/20200601173617101.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Noa3N0YXJ0,size_16,color_FFFFFF,t_70
[20200601173648879.png]: https://img-blog.csdnimg.cn/20200601173648879.png