消息队列之ActiveMQ

逃离我推掉我的手 2022-09-28 13:19 284阅读 0赞

一.什么是ActiveMQ?

ActiveMQ 是Apache出品，最流行的，能力强劲的开源消息总线。ActiveMQ 是一个完全支持JMS1.1和J2EE 1.4规范的 JMS Provider实现,尽管JMS规范出台已经是很久的事情了,但是JMS在当今的J2EE应用中间仍然扮演着特殊的地位。ActiveMQ可用在分布式服务的架构中,对服务之间进行解耦,是的业务拆分更加明确.

二.什么是JMS?

JMS是Java消息服务,定义了接口规范,提供了五种不同的消息正文格式，以及调用的消息类型，允许你发送并接收以一些不同形式的数据，提供现有消息格式的一些级别的兼容性。

·  StreamMessage -- Java原始值的数据流

·  MapMessage--一套名称\-值对

·  TextMessage--一个字符串对象

·  ObjectMessage--一个序列化的 Java对象

·  BytesMessage--一个字节的数据流

三.ActiveMQ的消息形式

对于消息的传递有两种类型：

**点对点模式:**即一个生产者和一个消费者一一对应；

**发布/订阅模式**:即一个生产者产生消息并进行发送后，可以由多个消费者进行接收。

四.ActiveMQ在linux系统上的安装

1.下载,网址是http://activemq.apache.org/download.html([进入下载页面][Link 1])

![Center][]

2.在linux系统上安装,需要jdk的生产环境,所以要在linux系统上安装好JDK(之前博客有说明怎么安装)

3.安装步骤:下载的ActiveMQ压缩包上传到linux系统上--->解压缩--->进入bin目录下启动ActiveMQ

进入bin目录后,其中命令./activemq start是启动,命令./activemq stop是关闭,命令./activemq status是查看状态.

4.进入ActiveMQ的管理后台(http://ip:8161/admin),ip是你安装ActiveMQ的linux系统的ip地址

![Center 1][]

![Center 2][]

五:ActiveMQ的使用方法

1.点对点模式和发布/订阅模式:

![Center 3][]

2.点对点模式(queue)之生产者(Producer),把对应版本的ActiveMQ的jar加入到工程中

public void testQueueProducer() throws Exception {
    		// 第一步：创建ConnectionFactory对象，需要指定服务端ip及端口号。,ActiveMQ传输协议是TCP协议
    		ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://ip:61616");
    		// 第二步：使用ConnectionFactory对象创建一个Connection对象。
    		Connection connection = connectionFactory.createConnection();
    		// 第三步：开启连接，调用Connection对象的start方法。
    		connection.start();
    		// 第四步：使用Connection对象创建一个Session对象。
    		//第一个参数：是否开启事务。true：开启事务，第二个参数忽略。
    		//第二个参数：当第一个参数为false时，才有意义。

//消息的应答模式。1、自动应答2、手动应答。一般是自动应答。
    		Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    		// 第五步：使用Session对象创建一个Destination对象（topic、queue），此处创建一个Queue对象。
    		//参数：队列的名称。
    		Queue queue = session.createQueue("zjl-queue");
    		// 第六步：使用Session对象创建一个Producer对象。
    		MessageProducer producer = session.createProducer(queue);
    		// 第七步：创建一个Message对象，创建一个TextMessage对象。
    		TextMessage textMessage = session.createTextMessage("hello activeMq!");
    		// 第八步：使用Producer对象发送消息。
    		producer.send(textMessage);
    		// 第九步：关闭资源。
    		producer.close();
    		session.close();
    		connection.close();
    	}

点对点模式之消费者(Consumer)

public void testQueueConsumer() throws Exception {
    		// 第一步：创建一个ConnectionFactory对象。
    		ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://ip:61616");
    		// 第二步：从ConnectionFactory对象中获得一个Connection对象。
    		Connection connection = connectionFactory.createConnection();
    		// 第三步：开启连接。调用Connection对象的start方法。
    		connection.start();
    		// 第四步：使用Connection对象创建一个Session对象。
    		Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    		// 第五步：使用Session对象创建一个Destination对象。和发送端保持一致queue，并且队列的名称一致。
    		Queue queue = session.createQueue("zjl-queue");
    		// 第六步：使用Session对象创建一个Consumer对象。
    		MessageConsumer consumer = session.createConsumer(queue);
    		// 第七步：接收消息。
    		consumer.setMessageListener(new MessageListener() {
    			
    			@Override
    			public void onMessage(Message message) {
    				try {
    					TextMessage textMessage = (TextMessage) message;
    					String text = null;
    					//取消息的内容
    					text = textMessage.getText();
    					// 第八步：打印消息。
    					System.out.println(text);
    				} catch (JMSException e) {
    					e.printStackTrace();
    				}
    			}
    		});
    		// 第九步：关闭资源
    		consumer.close();
    		session.close();
    		connection.close();
    	}

3.发布/订阅模式(Topic)之生产者(Producer)

public void testTopicProducer() throws Exception {
    		// 第一步：创建ConnectionFactory对象，需要指定服务端ip及端口号。
    		ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://ip:61616");
    		// 第二步：使用ConnectionFactory对象创建一个Connection对象。
    		Connection connection = connectionFactory.createConnection();
    		// 第三步：开启连接，调用Connection对象的start方法。
    		connection.start();
    		// 第四步：使用Connection对象创建一个Session对象。参数定义和queue是一样的
    		Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    		// 第五步：使用Session对象创建一个Destination对象（topic、queue），此处创建一个topic对象。
    		// 参数：话题的名称。
    		Topic topic = session.createTopic("zjl-topic");
    		// 第六步：使用Session对象创建一个Producer对象。
    		MessageProducer producer = session.createProducer(topic);
    		// 第七步：创建一个Message对象，创建一个TextMessage对象。
    		TextMessage textMessage = session.createTextMessage("hello activeMq!");
    		// 第八步：使用Producer对象发送消息。
    		producer.send(textMessage);
    		// 第九步：关闭资源。
    		producer.close();
    		session.close();
    		connection.close();
    	}

发布/订阅模式(Topic)之消费者(Consumer)

public void testTopicConsumer() throws Exception {
    		// 第一步：创建一个ConnectionFactory对象。
    		ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://ip:61616");
    		// 第二步：从ConnectionFactory对象中获得一个Connection对象。
    		Connection connection = connectionFactory.createConnection();
    		// 第三步：开启连接。调用Connection对象的start方法。
    		connection.start();
    		// 第四步：使用Connection对象创建一个Session对象。
    		Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    		// 第五步：使用Session对象创建一个Destination对象。和发送端保持一致topic，并且话题的名称一致。
    		Topic topic = session.createTopic("zjl-topic");
    		// 第六步：使用Session对象创建一个Consumer对象。
    		MessageConsumer consumer = session.createConsumer(topic);
    		// 第七步：接收消息。
    		consumer.setMessageListener(new MessageListener() {
    
    			@Override
    			public void onMessage(Message message) {
    				try {
    					TextMessage textMessage = (TextMessage) message;
    					String text = null;
    					// 取消息的内容
    					text = textMessage.getText();
    					// 第八步：打印消息。
    					System.out.println(text);
    				} catch (JMSException e) {
    					e.printStackTrace();
    				}
    			}
    		});
    		// 第九步：关闭资源
    		consumer.close();
    		session.close();
    		connection.close();
    	}

注:点对点模式是当生产的消息被消费掉时,该消息就不存在了,发布/订阅模式是生产的消息,谁订阅谁消费

六.Spring框架整合ActiveMQ

1.引入activemq-all.spring-jms.spring-context-support的jar包

2.配置文件中配置ConnectionFactory

<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
    		<property name="brokerURL" value="tcp://ip:61616" />
    	</bean>
    	
    	<bean id="connectionFactory"
    		class="org.springframework.jms.connection.SingleConnectionFactory">
    		<property name="targetConnectionFactory" ref="targetConnectionFactory" />
    	</bean>
     
    	
    	<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
    		
    		<property name="connectionFactory" ref="connectionFactory" />
    	</bean>
    	
    	<bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">
    		<constructor-arg>
    			<value>spring-queue</value>
    		</constructor-arg>
    	</bean>
    	
    	<bean id="topicDestination" class="org.apache.activemq.command.ActiveMQTopic">
    		<constructor-arg value="topic" />
    	</bean>

3.代码测试

发送消息:

public void testQueueProducer() throws Exception {
    		// 第一步：初始化一个spring容器,xml文件是配置ActiveMQ的那个文件
    		ApplicationContext applicationContext =

new ClassPathXmlApplicationContext("classpath:spring/applicationContext.xml");
    		// 第二步：从容器中获得JMSTemplate对象。
    		JmsTemplate jmsTemplate = applicationContext.getBean(JmsTemplate.class);
    		// 第三步：从容器中获得一个Destination对象
    		Queue queue = (Queue) applicationContext.getBean("queueDestination");
    		// 第四步：使用JMSTemplate对象发送消息，需要知道Destination
    		jmsTemplate.send(queue, new MessageCreator() {
    			
    			@Override
    			public Message createMessage(Session session) throws JMSException {
    				TextMessage textMessage = session.createTextMessage("spring activemq test");
    				return textMessage;
    			}
    		});
    	}

接受消息:

public class MyMessageListener implements MessageListener {//需要创建MessageListener的一个实现类
    	 @Override
    	 public void onMessage(Message message) {
    		 try {
    			 TextMessage textMessage = (TextMessage) message;
    			 //取消息内容
    			 String text = textMessage.getText();
    			 System.out.println(text);
    		 } catch (JMSException e) {
    		 	e.printStackTrace();
    		 }
    	 }
    
     }

[Link 1]: http://activemq.apache.org/download.html
[Center]: /images/20220708/3822c284a8894ef180749b61fa36cbc0.png
[Center 1]: /images/20220708/1e4980c116a54d4f82b2d9e842848643.png
[Center 2]: /images/20220708/7f84ee9a67cb40e6af2e381308f6ccf0.png
[Center 3]: /images/20220708/cb774af0add646928841bbed623ecd36.png