ActiveMQ常见消息类型

　　JMS由下面三部分组成:消息头、属性、消息体。其中消息体定义了五种消息体格式，也可以称为消息类型。

　　JMS规范中的消息类型包括TextMessage、MapMessage、ObjectMessage、BytesMessage、和StreamMessage等五种。ActiveMQ也有对应的实现，下面我们结合Spring JMS分别来看一下五种消息类型的收发代码。类结构如下图:

　　下面结合spring研究常见的几种消息类型:

applicationContext.xml如下：

1<?xml version="1.0" encoding="UTF-8"?>
2<beans xmlns="http://www.springframework.org/schema/beans"
3       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
4       xmlns:context="http://www.springframework.org/schema/context"
5       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">
6    <!--配置扫描注解的包-->
7    <context:component-scan base-package="cn.qlq.jms"></context:component-scan>
8
9    <!--Spring为我们提供的ConnectionFactory-->
10    <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
11        <property name="brokerURL" value="tcp://localhost:61616"></property>
12    </bean>
13
14    <!--Spring jms为我们提供的连接池-->
15    <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
16        <property name="targetConnectionFactory" ref="targetConnectionFactory"/>
17    </bean>
18
19    <!--一个队列的目的地，构造方法指定queueName，点对点模式-->
20    <bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">
21        <constructor-arg value="queue"></constructor-arg>
22    </bean>
23
24    <!--spring JMS提供的JmsTemplate-->
25    <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
26        <property name="connectionFactory" ref="connectionFactory"/>
27    </bean>
28
29
30    <!--消费消息容器(需要注入连接工厂，目的地，消息消费者监听器)-->
31    <bean id="jmsContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
32        <property name="connectionFactory" ref="connectionFactory"/>
33        <property name="destination" ref="queueDestination"/>
34        <property name="messageListener" ref="myConsumer"/>
35    </bean>
36
37    <!--一个队列的目的地，构造方法指定queueName，点对点模式-->
38    <bean id="topicDestination" class="org.apache.activemq.command.ActiveMQTopic">
39        <constructor-arg value="topic"></constructor-arg>
40    </bean>
41</beans>

1.最常见的  TextMessage

1public void sendMessage(final String message) {
2        jmsTemplate.send(destination, new MessageCreator() {
3            public Message createMessage(Session session) throws JMSException {
4                TextMessage textMessage = session.createTextMessage(message);
5                return textMessage;
6            }
7        });
8        logger.info("send textMesage:{}",message);
9    }

 查看后台:

2.MapMessage发送Map消息

1public void sendMessage(final String message) {
2        jmsTemplate.send(destination, new MessageCreator() {
3            public Message createMessage(Session session) throws JMSException {
4                MapMessage mapMessage = session.createMapMessage();
5                mapMessage.setString("str",message);
6                mapMessage.setBoolean("boolean",false);
7                return mapMessage;
8            }
9        });
10    }

查看后台消息:

3.ObjectMessage发送序列化对象消息，Object必须实现Serializable接口

1public void sendMessage() {
2        jmsTemplate.send(destination, new MessageCreator() {
3            public Message createMessage(Session session) throws JMSException {
4                ObjectMessage objectMessage = session.createObjectMessage();
5                ArrayList<String> obj = new ArrayList();
6                obj.add("123");
7                obj.add("456");
8                objectMessage.setObject(obj);
9                return objectMessage;
10            }
11        });
12    }

 查看后台消息:

4.  bytesMessage 发送字节消息

1public void sendMessage() {
2        jmsTemplate.send(destination, new MessageCreator() {
3            public Message createMessage(Session session) throws JMSException {
4                BytesMessage bytesMessage = session.createBytesMessage();
5                bytesMessage.writeBytes("str".getBytes());
6                bytesMessage.writeBoolean(false);
7                return bytesMessage;
8            }
9        });
10    }

后台:

5.StreamMessage　　发送Stream消息

1public void sendMessage() {
2        jmsTemplate.send(destination, new MessageCreator() {
3            public Message createMessage(Session session) throws JMSException {
4                StreamMessage streamMessage = session.createStreamMessage();
5                streamMessage.writeString("str");
6                streamMessage.writeInt(100);
7                streamMessage.writeObject("obj");
8                return streamMessage;
9            }
10        });
11    }

消息的接受者:

1public void onMessage(Message message) {
2        try{
3            System.out.println(message.getClass());
4
5            // TextMessage
6            if (message instanceof TextMessage) {
7                TextMessage textMessage1 = (TextMessage) message;
8                System.out.println(textMessage1.getText());
9            }
10
11            // MapMessage
12            if (message instanceof MapMessage) {
13                MapMessage mapMessage = (MapMessage) message;
14                System.out.println(mapMessage.toString());
15            }
16
17            // ObjectMessage
18            if (message instanceof ObjectMessage) {
19                ObjectMessage objectMessage = (ObjectMessage) message;
20                System.out.println(objectMessage.toString());
21            }
22
23            // BytesMessage
24            if (message instanceof BytesMessage) {
25                byte[] b = new byte[1024];
26                int len = -1;
27                BytesMessage bm = (BytesMessage) message;
28                while ((len = bm.readBytes(b)) != -1) {
29                    System.out.println(new String(b, 0, len));
30                }
31            }
32
33            // StreamMessage
34            if (message instanceof StreamMessage) {
35                StreamMessage sm = (StreamMessage) message;
36                System.out.println(sm.readString());
37                System.out.println(sm.readInt());
38            }
39        }catch (Exception e){
40
41        }
42    }