JMS由下面三部分组成:消息头、属性、消息体。其中消息体定义了五种消息体格式,也可以称为消息类型。
JMS规范中的消息类型包括TextMessage、MapMessage、ObjectMessage、BytesMessage、和StreamMessage等五种。ActiveMQ也有对应的实现,下面我们结合Spring JMS分别来看一下五种消息类型的收发代码。类结构如下图:
下面结合spring研究常见的几种消息类型:
applicationContext.xml如下:
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:context="http://www.springframework.org/schema/context"
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.qlq.jms"></context:component-scan>
<!--Spring为我们提供的ConnectionFactory-->
<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="tcp://localhost:61616"></property>
</bean>
<!--Spring jms为我们提供的连接池-->
<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
<property name="targetConnectionFactory" ref="targetConnectionFactory"/>
</bean>
<!--一个队列的目的地,构造方法指定queueName,点对点模式-->
<bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg value="queue"></constructor-arg>
</bean>
<!--spring JMS提供的JmsTemplate-->
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<property name="connectionFactory" ref="connectionFactory"/>
</bean>
<!--消费消息容器(需要注入连接工厂,目的地,消息消费者监听器)-->
<bean id="jmsContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="connectionFactory"/>
<property name="destination" ref="queueDestination"/>
<property name="messageListener" ref="myConsumer"/>
</bean>
<!--一个队列的目的地,构造方法指定queueName,点对点模式-->
<bean id="topicDestination" class="org.apache.activemq.command.ActiveMQTopic">
<constructor-arg value="topic"></constructor-arg>
</bean>
</beans>
1.最常见的 TextMessage
public void sendMessage(final String message) {
jmsTemplate.send(destination, new MessageCreator() {
public Message createMessage(Session session) throws JMSException {
TextMessage textMessage = session.createTextMessage(message);
return textMessage;
}
});
logger.info("send textMesage:{}",message);
}
查看后台:
2.MapMessage发送Map消息
public void sendMessage(final String message) {
jmsTemplate.send(destination, new MessageCreator() {
public Message createMessage(Session session) throws JMSException {
MapMessage mapMessage = session.createMapMessage();
mapMessage.setString("str",message);
mapMessage.setBoolean("boolean",false);
return mapMessage;
}
});
}
查看后台消息:
3.ObjectMessage发送序列化对象消息,Object必须实现Serializable接口
public void sendMessage() {
jmsTemplate.send(destination, new MessageCreator() {
public Message createMessage(Session session) throws JMSException {
ObjectMessage objectMessage = session.createObjectMessage();
ArrayList<String> obj = new ArrayList();
obj.add("123");
obj.add("456");
objectMessage.setObject(obj);
return objectMessage;
}
});
}
查看后台消息:
4. bytesMessage 发送字节消息
public void sendMessage() {
jmsTemplate.send(destination, new MessageCreator() {
public Message createMessage(Session session) throws JMSException {
BytesMessage bytesMessage = session.createBytesMessage();
bytesMessage.writeBytes("str".getBytes());
bytesMessage.writeBoolean(false);
return bytesMessage;
}
});
}
后台:
5.StreamMessage 发送Stream消息
public void sendMessage() {
jmsTemplate.send(destination, new MessageCreator() {
public Message createMessage(Session session) throws JMSException {
StreamMessage streamMessage = session.createStreamMessage();
streamMessage.writeString("str");
streamMessage.writeInt(100);
streamMessage.writeObject("obj");
return streamMessage;
}
});
}
消息的接受者:
public void onMessage(Message message) {
try{
System.out.println(message.getClass());
// TextMessage
if (message instanceof TextMessage) {
TextMessage textMessage1 = (TextMessage) message;
System.out.println(textMessage1.getText());
}
// MapMessage
if (message instanceof MapMessage) {
MapMessage mapMessage = (MapMessage) message;
System.out.println(mapMessage.toString());
}
// ObjectMessage
if (message instanceof ObjectMessage) {
ObjectMessage objectMessage = (ObjectMessage) message;
System.out.println(objectMessage.toString());
}
// BytesMessage
if (message instanceof BytesMessage) {
byte[] b = new byte[1024];
int len = -1;
BytesMessage bm = (BytesMessage) message;
while ((len = bm.readBytes(b)) != -1) {
System.out.println(new String(b, 0, len));
}
}
// StreamMessage
if (message instanceof StreamMessage) {
StreamMessage sm = (StreamMessage) message;
System.out.println(sm.readString());
System.out.println(sm.readInt());
}
}catch (Exception e){
}
}