消息中间件解决方案ActiveMQ

ゞ 浴缸里的玫瑰 2023-02-14 07:43 67阅读 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应用中间仍然扮演着特殊的地位。
对于消息的传递有两种类型：
一种是点对点的，即一个生产者和一个消费者一一对应；
在这里插入图片描述
另一种是发布/ 订阅模式，即一个生产者产生消息并进行发送后，可以由多个消费者进
行接收。
在这里插入图片描述
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）解压此文件

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

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

  1. chmod 777 apache-activemq-5.12.0

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

  1. 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启动

  1. ./activemq start

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

1.5 ActiveMQ入门小DEMO（点对点）

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

1.5.1消息生产者

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

  1. <dependency>
  2.               <groupId>org.apache.activemq</groupId>
  3.               <artifactId>activemq-all</artifactId>
  4.               <version>5.11.2</version>
  5.           </dependency>

（2）创建类QueueProducer

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

上述代码中第4步创建session 的两个参数：
第1个参数 是否使用事务
第2个参数 消息的确认模式
AUTO_ACKNOWLEDGE = 1 自动确认
CLIENT_ACKNOWLEDGE = 2 客户端手动确认
DUPS_OK_ACKNOWLEDGE = 3 自动批量确认
SESSION_TRANSACTED = 0 事务提交并确认
运行后通过ActiveMQ管理界面查询
在这里插入图片描述

1.5.2消息消费者

创建类

  1. package cn.atguigu.demo;
  2. import java.io.IOException;
  3. import javax.jms.Connection;
  4. import javax.jms.JMSException;
  5. import javax.jms.Message;
  6. import javax.jms.MessageConsumer;
  7. import javax.jms.MessageListener;
  8. import javax.jms.Queue;
  9. import javax.jms.Session;
  10. import javax.jms.TextMessage;
  11. import org.apache.activemq.ActiveMQConnectionFactory;
  12. /** * 消息消费者 * @author Administrator * */
  13. public class QueueConsumer { 
  14.     public static void main(String[] args) throws JMSException, IOException { 
  15.         //1.创建连接工厂
  16.         ActiveMQConnectionFactory connectionFactory=new ActiveMQConnectionFactory("tcp://192.168.25.129:61616");        
  17.         //2.获取连接
  18.         Connection connection = connectionFactory.createConnection();
  19.         //3.启动连接
  20.         connection.start();
  21.         //4.获取session
  22.         Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
  23.         //5.创建队列对象
  24.         Queue queue = session.createQueue("test-queue");
  25.         //6.创建消息消费者
  26.         MessageConsumer consumer=session.createConsumer(queue);
  27.         //7.接受消息
  28.         consumer.setMessageListener(new MessageListener() {             
  29.             @Override
  30.             public void onMessage(Message message) { 
  31.                 //获取文本消息对象
  32.                 TextMessage textMessage=(TextMessage)message;                
  33.                 try { 
  34.                     String text = textMessage.getText();//提取文本
  35.                     System.out.println(text);
  36.                 } catch (JMSException e) { 
  37.                     // TODO Auto-generated catch block
  38.                     e.printStackTrace();
  39.                 }                
  40.             }
  41.         });
  42.         //8.等待键盘输入
  43.         System.in.read();
  44.         //9.关闭资源 
  45.         consumer.close();
  46.         session.close();
  47.         connection.close();
  48.     }
  49. }

执行后看到控制台输出

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

1.6.1消息生产者

创建类

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

运行效果如下：
在这里插入图片描述

1.6.2消息消费者

代码：

  1. package cn.atguigu.demo;
  2. import java.io.IOException;
  3. import javax.jms.Connection;
  4. import javax.jms.JMSException;
  5. import javax.jms.Message;
  6. import javax.jms.MessageConsumer;
  7. import javax.jms.MessageListener;
  8. import javax.jms.Queue;
  9. import javax.jms.Session;
  10. import javax.jms.TextMessage;
  11. import javax.jms.Topic;
  12. import org.apache.activemq.ActiveMQConnectionFactory;
  13. public class TopicConsumer { 
  14.     public static void main(String[] args) throws JMSException, IOException { 
  15.         //1.创建连接工厂
  16.         ActiveMQConnectionFactory connectionFactory=new ActiveMQConnectionFactory("tcp://192.168.25.129:61616");        
  17.         //2.获取连接
  18.         Connection connection = connectionFactory.createConnection();
  19.         //3.启动连接
  20.         connection.start();
  21.         //4.获取session
  22.         Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
  23.         //5.创建主题对象
  24.         Topic topic = session.createTopic("test-topic");
  25.         //6.创建消息消费者
  26.         MessageConsumer consumer=session.createConsumer(topic);
  27.         //7.接受消息
  28.         consumer.setMessageListener(new MessageListener() {             
  29.             @Override
  30.             public void onMessage(Message message) { 
  31.                 //获取文本消息对象
  32.                 TextMessage textMessage=(TextMessage)message;                
  33.                 try { 
  34.                     String text = textMessage.getText();//提取文本
  35.                     System.out.println(text);
  36.                 } catch (JMSException e) { 
  37.                     // TODO Auto-generated catch block
  38.                     e.printStackTrace();
  39.                 }                
  40.             }
  41.         });
  42.         //8.等待键盘输入
  43.         System.in.read();
  44.         //9.关闭资源 
  45.         consumer.close();
  46.         session.close();
  47.         connection.close();
  48.     }
  49. }

发表评论

表情：
评论列表 （有 0 条评论，67人围观）

还没有评论，来说两句吧...

相关阅读

    相关 消息中间件（ActiveMQ）

    原文地址：[手把手教你如何玩转消息中间件（ActiveMQ）][ActiveMQ] 情景引入 小白：起床起床起床起床。。。。快起床~ 我：怎么了又，大惊小怪，吓到我

    冷不防冷不防/ 0 赞/ 554 阅读