消息中间件-activeMQ

时间 2019-12-06

原文原文链接

JMS简介

全称：Java Message Service 中文：Java消息服务。
JMS是Java的一套API标准，最初的目的是为了使应用程序可以访问现有的MOM系统（MOM是Message Oriented Middleware的英文缩写，指的是利用高效可靠的消息传递机制进行平台无关的数据交流，并基于数据通讯来进行分布式系统的集成。）；后来被许多现有的MOM供应商采用，并实现为MOM系统。
基于JMS实现的MOM，又被称为JMS Provider。
html

经常使用的消息中间件有哪些
1ActiveMQ
ActiveMQ 是Apache出品，最流行的，能力强劲的开源消息总线。ActiveMQ 是一个彻底支持JMS1.1和J2EE 1.4规范的 JMS Provider实现。java

2RabbitMQ
RabbitMQ是一个在AMQP基础上完成的，可复用的企业消息系统。他遵循Mozilla Public License开源协议。开发语言为Erlang。spring

3RocketMQ
由阿里巴巴定义开发的一套消息队列应用服务。
apache

在用activeMQ获取对象(objectmessage)数据消息时 ,若是显示以下错误(这里在activeMQ里取到了消息,可是打印不出来):缓存

Exception in thread "main" javax.jms.JMSException: Failed to build body from content. Serializable class not available to broker. Reason: java.lang.ClassNotFoundException: Forbidden class com.liy.pojo.User! This class is not trusted to be serialized as ObjectMessage payload. Please take a look at http://activemq.apache.org/objectmessage.html for more information on how to configure trusted classes.
   at org.apache.activemq.util.JMSExceptionSupport.create(JMSExceptionSupport.java:36)
   at org.apache.activemq.command.ActiveMQObjectMessage.getObject(ActiveMQObjectMessage.java:208)
   at com.liy.comsumer.ObjectConsumer.main(ObjectConsumer.java:39)
Caused by: java.lang.ClassNotFoundException: Forbidden class com.liy.pojo.User! This class is not trusted to be serialized as ObjectMessage payload. Please take a look at http://activemq.apache.org/objectmessage.html for more information on how to configure trusted classes.
   at org.apache.activemq.util.ClassLoadingAwareObjectInputStream.checkSecurity(ClassLoadingAwareObjectInputStream.java:112)
   at org.apache.activemq.util.ClassLoadingAwareObjectInputStream.resolveClass(ClassLoadingAwareObjectInputStream.java:57)
   at java.io.ObjectInputStream.readNonProxyDesc(Unknown Source)
   at java.io.ObjectInputStream.readClassDesc(Unknown Source)
   at java.io.ObjectInputStream.readOrdinaryObject(Unknown Source)
   at java.io.ObjectInputStream.readObject0(Unknown Source)
   at java.io.ObjectInputStream.readObject(Unknown Source)
   at org.apache.activemq.command.ActiveMQObjectMessage.getObject(ActiveMQObjectMessage.java:206)
   ... 1 more
session

说找不到能够信任的实体类(pojo类)什么的 app

那么让他信任全部包下的类框架

//设置全部包下的类都是能够信任的
((ActiveMQConnectionFactory) factory).setTrustAllPackages(true);maven

activeMQ和spring的使用tcp

建立两个maven的jar工程一个用来发送消息 ,一个用来接收消息

Activemq-02-spring-producer工程里步骤

1.引入依赖(另一个工程里依赖是同样的 , 这生产者和消费者彻底能够写在一个工程了)

<dependencies>
	<!-- ActiveMQ客户端完整jar包依赖 -->
	<dependency>
		<groupId>org.apache.activemq</groupId>
		<artifactId>activemq-all</artifactId>
		<version>5.9.0</version>
	</dependency>
	<!-- ActiveMQ和Spring整合配置文件标签处理jar包依赖 -->
	<dependency>
		<groupId>org.apache.xbean</groupId>
		<artifactId>xbean-spring</artifactId>
		<version>4.5</version>
	</dependency>
	<!-- Spring-JMS插件相关jar包依赖 -->
	<dependency>
		<groupId>org.springframework</groupId>
		<artifactId>spring-jms</artifactId>
		<version>4.1.6.RELEASE</version>
	</dependency>
	<!-- Spring框架上下文jar包依赖 -->
	<dependency>
		<groupId>org.springframework</groupId>
		<artifactId>spring-context</artifactId>
		<version>4.1.6.RELEASE</version>
	</dependency>
</dependencies>

2.随便写个pojo类,用来传输的数据消息(注意必须实现序列化,以及写上这个序列UID)

public class User implements Serializable{
 
	private static final long serialVersionUID = 1L;
	
	private String name;
	
	private String pwd;

3.写生产类

package com.liy.producer;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.ObjectMessage;
import javax.jms.Session;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import com.liy.pojo.User;

public class Producers {
	
	private JmsTemplate template;
	
	public JmsTemplate getTemplate() {
		return template;
	}

	public void setTemplate(JmsTemplate template) {
		this.template = template;
	}

	/**
	 * 发送消息
	 * @param destinationName 
	 */
	public void send(String destinationName,User user){
		template.send(destinationName, new MessageCreator() {
			
			@Override
			public Message createMessage(Session arg0) throws JMSException {
				 ObjectMessage message = arg0.createObjectMessage(user);
				return message;
			}
		});
		
	}
	
}

4.写spring的配置文件(applicationContext.xml)

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
	xmlns:context="http://www.springframework.org/schema/context"
	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" 
	xmlns:amq="http://activemq.apache.org/schema/core"
	xmlns:jms="http://www.springframework.org/schema/jms"
	xsi:schemaLocation="http://www.springframework.org/schema/beans     
        http://www.springframework.org/schema/beans/spring-beans-4.0.xsd     
        http://www.springframework.org/schema/context     
        http://www.springframework.org/schema/context/spring-context-4.0.xsd  
        http://www.springframework.org/schema/jms  
        http://www.springframework.org/schema/jms/spring-jms-4.0.xsd  
        http://activemq.apache.org/schema/core  
        http://activemq.apache.org/schema/core/activemq-core-5.9.0.xsd">
        
    <!-- 添加扫描 -->
	<!--<context:component-scan base-package="com.liy.*"></context:component-scan>  -->
	
	<!-- ActiveMQ 链接工厂 -->
	<!-- 真正能够产生Connection的ConnectionFactory,由对应的 JMS服务厂商提供 -->
	<!-- 需提供访问路径tcp://ip:61616;以及用户名,密码 -->
	<amq:connectionFactory id="amqConnectionFactory"
		brokerURL="tcp://192.168.72.114:61616" userName="admin" password="admin" />

	<!-- Spring Caching链接工厂 -->
	<!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory -->
	<bean id="connectionFactory"
		class="org.springframework.jms.connection.CachingConnectionFactory">
		<!-- 目标ConnectionFactory对应真实的能够产生JMS Connection的ConnectionFactory -->
		<property name="targetConnectionFactory" ref="amqConnectionFactory"></property>
		<!-- Session缓存数量 -->
		<property name="sessionCacheSize" value="100" />
	</bean>

	<!-- 消息生产者 start -->

	<!-- 定义JmsTemplate对象. 此类型由Spring框架JMS组件提供. 用于访问ActiveMQ使用. -->
	<bean id="jmsQueueTemplate" class="org.springframework.jms.core.JmsTemplate">
		<!-- 这个connectionFactory对应的是咱们定义的Spring提供的那个ConnectionFactory对象 -->
		<constructor-arg ref="connectionFactory" />
		<!-- 非pub/sub模型（发布/订阅）,即队列模式, 默认数据可省略配置 -->
		<!-- <property name="pubSubDomain" value="false" /> -->
	</bean>
	
	<!-- 定义生成者对象 -->
	<bean id="Producer" class="com.liy.producer.Producers">
		<!-- 为属性赋值 -->
		<property name="template" ref="jmsQueueTemplate"></property>
	</bean>
	
	 
</beans>

5.写生产工程的测试类

public class Test {
	public static void main(String[] args) {
		ApplicationContext ac = new ClassPathXmlApplicationContext("applicationContext.xml");
		Producers pro = ac.getBean(Producers.class);
	
		User user = new User("liy","666");
		
		pro.send("act-spring", user);
		System.out.println("消息发送完成....");
	}
}

以下即为启动成功,消息发送完了 , 这里就算关掉也不要紧

消费者工程步骤

1.依赖和生产者依赖同样

2.实体类也是同样的,复制下来便可

3.接下来写消费者类

package com.liy.consumer;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.ObjectMessage;

public class Consumers implements MessageListener{

	@Override
	public void onMessage(Message message) {
		 ObjectMessage obj = (ObjectMessage) message;
		try {
			System.out.println(obj.getObject());
		} catch (JMSException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	
	
}

4.写spring的配置文件

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
	xmlns:context="http://www.springframework.org/schema/context"
	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" 
	xmlns:amq="http://activemq.apache.org/schema/core"
	xmlns:jms="http://www.springframework.org/schema/jms"
	xsi:schemaLocation="http://www.springframework.org/schema/beans     
        http://www.springframework.org/schema/beans/spring-beans-4.0.xsd     
        http://www.springframework.org/schema/context     
        http://www.springframework.org/schema/context/spring-context-4.0.xsd  
        http://www.springframework.org/schema/jms  
        http://www.springframework.org/schema/jms/spring-jms-4.0.xsd  
        http://activemq.apache.org/schema/core  
        http://activemq.apache.org/schema/core/activemq-core-5.9.0.xsd">
    <!-- 添加扫描 -->
	<!-- <context:component-scan base-package="com.liy.*"></context:component-scan>-->
	
	<!-- ActiveMQ 链接工厂 -->
	<!-- 真正能够产生Connection的ConnectionFactory,由对应的 JMS服务厂商提供 -->
	<!-- 需提供访问路径tcp://ip:61616;以及用户名,密码 -->
	<amq:connectionFactory id="amqConnectionFactory"
		brokerURL="tcp://192.168.72.114:61616" userName="admin" password="admin" />

	<!-- Spring Caching链接工厂 -->
	<!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory -->
	<bean id="connectionFactory"
		class="org.springframework.jms.connection.CachingConnectionFactory">
		<!-- 目标ConnectionFactory对应真实的能够产生JMS Connection的ConnectionFactory -->
		<property name="targetConnectionFactory" ref="amqConnectionFactory"></property>
		<!-- Session缓存数量 -->
		<property name="sessionCacheSize" value="100" />
	</bean>

	 

	<!-- 消息消费者 start -->

	<!-- 定义消息监听器, 此组件为spring-jms组件定义. 能够一次注册若干消息监听器.
		属性解释:
			destination-type - 目的地类型, queue表明消息队列
				可选值: queue | topic | durableTopic
				queue - 默认值. 表明消息队列
				topic - 表明消息队列集合
				durableTopic - 持久化的消息队列集合. ActiveMQ会保证消息的消费者必定接收到此消息.
			container-type - 容器类型
				可选值: default | simple
				default - 默认值. 默认容器类型, 对应DefaultMessageListenerContainer
				simple - 简单容器类型, 对应SimpleMessageListenerContainer
			connection-factory - 连接工厂, 注入的是Spring-JMS组件提供的连接工厂对象.
			acknowledge - 确认方式
				可选值: auto | client | dups-ok | transacted
				auto - 默认值, 即自动确认消息
				client - 客户端确认消息
				dups-ok - 可以使用副本的客户端确认消息
				transacted - 有事务的持久化消息确认机制. 需开启对ActiveMQ的事务控制才可应用. 
	 -->
	<jms:listener-container destination-type="queue"
		container-type="default" connection-factory="connectionFactory"
		acknowledge="auto">
		<!-- 注册消息监听器. 若是须要注册多个, 重复定义下述标签. -->
		<jms:listener destination="act-spring" ref="orderReciver" />
	</jms:listener-container>
	
	<!-- 容器管理消息监听器实现类对象 -->
	<bean id="orderReciver" class="com.liy.consumer.Consumers"/>

	<!-- 消息消费者 end -->
</beans>

5.消费工程的测试类(由于spring配置文件里有设置监听器 ,只要生产者工程发布了消息,监听器就会自动获取消息, 因此只要把spring容易开启便可)

public class Test {
	public static void main(String[] args) {
		ApplicationContext ac = new ClassPathXmlApplicationContext("applicationContext.xml");
	 
	}
}

如图