消息中间件

常用的消息中间件

1. ActiveMQ Apache出品的，开发常用
2. RabbitMQ 完全支持JMS1.1和J2EE 1.4规范
3. kafka Kafka的目的是通过Hadoop的并行加载机制来统一线上和离线的消息处理，也是为了通过集群来提供实时的消息。

消息中间件可以提高系统性能

JMS

JMS 是一套消息中间件的技术规范，定义了一系列的接口规范。

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

1. TextMessage–一个字符串对象
2. MapMessage–一套名称-值对
3. ObjectMessage–一个序列化的 Java 对象
4. BytesMessage–一个字节的数据流
5. StreamMessage – Java 原始值的数据流

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

1. 一种是点对点的，即一个生产者和一个消费者一一对应；

2. 另一种是发布、订阅模式，即一个生产者产生消息并进行发送后，可以由多个消费者进行接收。

   

消息中间件用的比较多的还是activeMQ

点对点模式

消息生产者生产消息（点对点模式）

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

public class QueueProducer {

    public static void main(String[] args) throws JMSException {
        //创建连接工厂
        ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory("tcp://192.168.13.131:61616");

        //创建连接
        Connection connection = activeMQConnectionFactory.createConnection();

        //启动连接
        connection.start();

        //获取session（会话对象） 参数1:是否启动事物 参数2:消息确认方式
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

        //创建队列对象
        Queue queue = session.createQueue("sms");

        //创建消息生产者对象
        MessageProducer producer = session.createProducer(queue);

        //创建消息(文本对象)
        TextMessage textMessage = session.createTextMessage("你好，jms");

        //发送消息
        producer.send(textMessage);

        //关闭连接
        producer.close();
        session.close();
        connection.close();

    }
}

消费者消费消息（点对点模式）

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

public class QueueConsumer {

    public static void main(String[] args) throws JMSException {
        //创建连接工厂
        ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory("tcp://192.168.13.131:61616");

        //创建连接
        Connection connection = activeMQConnectionFactory.createConnection();

        //启动连接
        connection.start();

        //获取session（会话对象） 参数1:是否启动事物 参数2:消息确认方式
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

        //创建队列对象
        Queue queue = session.createQueue("sms");

        //创建消息消费者对象
        MessageConsumer consumer = session.createConsumer(queue);

        //设置监听
        consumer.setMessageListener(new MessageListener() {
            public void onMessage(Message message) {
                TextMessage textMessage = (TextMessage) message;
                try {
                    String messageText = textMessage.getText();
                    System.out.println("接收到的消息是："+messageText);
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        });

        consumer.close();
        session.close();
        connection.close();

    }
}

发布订阅模式

消息生产者生产消息（发布订阅模式）

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

public class QueueProducer {

    public static void main(String[] args) throws JMSException {
        //创建连接工厂
        ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory("tcp://192.168.13.131:61616");

        //创建连接
        Connection connection = activeMQConnectionFactory.createConnection();

        //启动连接
        connection.start();

        //获取session（会话对象） 参数1:是否启动事物 参数2:消息确认方式
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

        //创建主题对象
        Topic topic = session.createTopic("sms2");


        //创建消息生产者对象
        MessageProducer producer = session.createProducer(topic);

        //创建消息(文本对象)
        TextMessage textMessage = session.createTextMessage("你好，jms2");

        //发送消息
        producer.send(textMessage);

        //关闭连接
        producer.close();
        session.close();
        connection.close();

    }
}

消费者消费消息（发布订阅模式）

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

public class QueueConsumer {

    public static void main(String[] args) throws JMSException {
        //创建连接工厂
        ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory("tcp://192.168.13.131:61616");

        //创建连接
        Connection connection = activeMQConnectionFactory.createConnection();

        //启动连接
        connection.start();

        //获取session（会话对象） 参数1:是否启动事物 参数2:消息确认方式
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

        //创建主题对象
        Topic topic = session.createTopic("sms2");

        //创建消息消费者对象
        MessageConsumer consumer = session.createConsumer(topic);

        //设置监听
        consumer.setMessageListener(new MessageListener() {
            public void onMessage(Message message) {
                TextMessage textMessage = (TextMessage) message;
                try {
                    String messageText = textMessage.getText();
                    System.out.println("接收到的消息是："+messageText);
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        });

        consumer.close();
        session.close();
        connection.close();

    }
}

注意点

• 发布订阅模式下，生产者发布消息，消费者必须在开启的情况下才可以接受到，否则消息就浪费掉，就消失了。

Spring整合Jms

Spring生产者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.xsd
		http://www.springframework.org/schema/context   
		http://www.springframework.org/schema/context/spring-context.xsd">
		
		
	<context:component-scan base-package="cn.itcast.demo"></context:component-scan>     
	
	   
    <!-- 真正可以产生Connection的ConnectionFactory，由对应的 JMS服务厂商提供-->  
	<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">  
	    <property name="brokerURL" value="tcp://192.168.13.131:61616"/>  
	</bean>
	   
    <!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory -->  
	<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">  
	<!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory -->  
	    <property name="targetConnectionFactory" ref="targetConnectionFactory"/>  
	</bean>  
		   
    <!-- Spring提供的JMS工具类，它可以进行消息发送、接收等 -->  
	<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">  
	    <!-- 这个connectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 -->  
	    <property name="connectionFactory" ref="connectionFactory"/>  
	</bean>      
    <!--这个是队列目的地，点对点的  文本信息-->  
	<bean id="queueTextDestination" class="org.apache.activemq.command.ActiveMQQueue">  
	    <constructor-arg value="queue_text"/>  
	</bean>    
	
	<!--这个是订阅模式  文本信息-->  
	<bean id="topicTextDestination" class="org.apache.activemq.command.ActiveMQTopic">  
	    <constructor-arg value="topic_text"/>  
	</bean>  
	
</beans>

Spring生产者生产消息（点对点）

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.stereotype.Component;

import javax.jms.Destination;

@Component
public class SpringQueueProducer {

    @Autowired
    private JmsTemplate jmsTemplate;

    @Autowired
    private Destination queueTextDestination;

    //发送消息
    public void sendTextMessage(final String text){
        jmsTemplate.convertAndSend(text);
    }
}

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations="classpath:applicationContext-jms-producer.xml")
public class TestQueue {

	@Autowired
	private QueueProducer queueProducer;
	
	@Test
	public void testSend(){
		queueProducer.sendTextMessage("SpringJms-点对点");
	}	
}

Spring消费者接收消息（点对点）
Spring消费者接收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.xsd
		http://www.springframework.org/schema/context   
		http://www.springframework.org/schema/context/spring-context.xsd">
	
    <!-- 真正可以产生Connection的ConnectionFactory，由对应的 JMS服务厂商提供-->  
	<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">  
	    <property name="brokerURL" value="tcp://192.168.25.131:61616"/>
	</bean>
	   
    <!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory -->  
	<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">  
	<!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory -->  
	    <property name="targetConnectionFactory" ref="targetConnectionFactory"/>  
	</bean>  
	
    <!--这个是队列目的地，点对点的  文本信息-->  
	<bean id="queueTextDestination" class="org.apache.activemq.command.ActiveMQQueue">  
	    <constructor-arg value="queue_text"/>  
	</bean>    
	
	<!-- 监听类 -->
	<bean id="myMessageListener" class="cn.itcast.demo.MyMessageListener"></bean>
	<!-- 消息监听容器 -->
	<bean class="org.springframework.jms.listener.DefaultMessageListenerContainer">
		<property name="connectionFactory" ref="connectionFactory" />
		<property name="destination" ref="queueTextDestination" />
		<property name="messageListener" ref="myMessageListener" />
	</bean>
	
</beans>

Spring消费者消费消息（点对点）

package cn.itcast.demo;

import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;

public class MyMessageListener implements MessageListener {


    public void onMessage(Message message) {
        TextMessage textMessage = (TextMessage) message;
        System.out.println("接受到的消息为："+textMessage);
    }
}

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations="classpath:applicationContext-jms-consumer-queue.xml")
public class TestQueue {
	@Test
	public void testQueue(){
		try {
			System.in.read();
		} catch (IOException e) {
			e.printStackTrace();
		}		
	}	
}

spring整合jms，对于订阅广播模式，xml配置和点对点一样，我被对应配置已经放在里面了
然后只有执行语句不一样

Spring生产者生产消息（发布订阅模式）

@Component
public class TopicProducer {
	@Autowired
	private JmsTemplate jmsTemplate;
	
	@Autowired
	private Destination topicTextDestination;
	
	/**
	 * 发送文本消息
	 * @param text
	 */
	public void sendTextMessage(final String text){
		jmsTemplate.send(topicTextDestination, new MessageCreator() {			
			public Message createMessage(Session session) throws JMSException {
				return session.createTextMessage(text);
			}
		});		
	}
}

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import cn.itcast.demo.TopicProducer;
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations="classpath:applicationContext-activemq-producer.xml")
public class TestTopic {
	@Autowired
	private TopicProducer topicProducer;
	@Test
	public void sendTextQueue(){		
		topicProducer.sendTextMessage();
	}	
}

Spring消费者消费消息（发布订阅模式）

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations="classpath:applicationContext-jms-consumer-topic.xml")
public class TestTopic {
	@Test
	public void testTopic(){
		try {
			System.in.read();
		} catch (IOException e) {
			e.printStackTrace();
		}		
	}		
}

上面的代码

System.in.read();

是防止程序启动结束就会关闭，所以让窗口一直等待输入

学习总结

花了一两天来了解jms以及和spring的整合，用处很大！有以下两点

• 可以用来在发送短信的功能中，把短信验证码发送到队列中，然后点对点方式让用户接受到，起到了一定程度的削锋
• 发布订阅模式可以用在集群环境下，由于集群环境下，同一个页面在不同的服务器上存在多份，可以使用发布订阅模式完成同时渲染多个服务器上的相同的页面