JMS消息组成详解
整个JMS协议组成结构如下
| 结构 | 描述 |
|---|---|
| JMS Provider | 消息中间件/消息服务器 |
| JMS Producer | 消息生产者 |
| JMS Consumer | 消息消费者 |
| JMS Message | 消息（重要） |
JMS Message消息由三部分组成：
1）消息头
2）消息体
3）消息属性
JMS消息头
JMS消息头预定义了若干字段用于客户端与JMS提供者之间识别和发送消息，预编译头如下：红色为重要的消息头
| 名称 | 描述 |
|---|---|
| JMSDestination | 消息发送的 Destination，在发送过程中由提供者设置 |
| JMSMessageID | 唯一标识提供者发送的每一条消息。这个字段是在发送过程中由提供者设 置的，客户机只能在消息发送后才能确定消息的 JMSMessageID |
| JMSDeliveryMode | 消息持久化。包含值 DeliveryMode.PERSISTENT 或者 DeliveryMode.NON_PERSISTENT。 |
| JMSTimestamp | 提供者发送消息的时间，由提供者在发送过程中设置 |
| JMSExpiration | 消息失效的时间，毫秒，值 0 表明消息不会过期，默认值为0 |
| JMSPriority | 消息的优先级，由提供者在发送过程中设置。优先级 0 的优先级最低，优 先级 9 的优先级最高。0-4为普通消息，5-9为加急消息。优先级高就一定先发送，只保证了加急消息必须先于普通消息发送。默认值为4 |
| JMSCorrelationID | 通常用来链接响应消息与请求消息，由发送消息的 JMS 程序设置。 |
| JMSReplyTo | 请求程序用它来指出回复消息应发送的地方，由发送消息的 JMS 程序设置 |
| JMSType | JMS 程序用它来指出消息的类型。 |
| JMSRedelivered | More Actions消息的重发标志，false，代表该消息是第一次发生，true，代表该消息为 重发消息 |
不过需要注意的是，在传送消息时，消息头的值由JMS提供者来设置，因此开发者使用以上setJMSXXX()方法分配的值就被忽略了，只有以下几个值是可以由开发者设置的：JMSCorrelationID，JMSReplyTo，JMSType
public static void main(String[] args) throws Exception{
ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
Connection connection = activeMQConnectionFactory.createConnection();
connection.start();
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Topic topic = session.createTopic(TOPIC_NAME);
MessageProducer messageProducer = session.createProducer(topic);
for (int i = 1; i < 4 ; i++) {
TextMessage textMessage = session.createTextMessage("topic_name--" + i);
// 这里可以指定每个消息的目的地
textMessage.setJMSDestination(topic);
/*
持久模式和非持久模式。
一条持久性的消息：应该被传送“一次仅仅一次”，这就意味着如果JMS提供者出现故障，该消息并不会丢失，它会在服务器恢复之后再次传递。
一条非持久的消息：最多会传递一次，这意味着服务器出现故障，该消息将会永远丢失。
*/
textMessage.setJMSDeliveryMode(0);
/*
可以设置消息在一定时间后过期，默认是永不过期。
消息过期时间，等于Destination的send方法中的timeToLive值加上发送时刻的GMT时间值。
如果timeToLive值等于0，则JMSExpiration被设为0，表示该消息永不过期。
如果发送后，在消息过期时间之后还没有被发送到目的地，则该消息被清除。
*/
textMessage.setJMSExpiration(1000);
/* 消息优先级，从0-9十个级别，0-4是普通消息5-9是加急消息。
JMS不要求MQ严格按照这十个优先级发送消息但必须保证加急消息要先于普通消息到达。默认是4级。
*/
textMessage.setJMSPriority(10);
// 唯一标识每个消息的标识。MQ会给我们默认生成一个，我们也可以自己指定。
textMessage.setJMSMessageID("ABCD");
// 上面有些属性在send方法里也能设置
messageProducer.send(textMessage);
}
messageProducer.close();
session.close();
connection.close();
System.out.println(" **** TOPIC_NAME消息发送到MQ完成 ****");
}
}
JMS消息体
在消息体中，JMS API定义了五种类型的消息格式，让我们可以以不同的形式发送和接受消息，并提供了对已有消息格式的兼容。不同的消息类型如下：
JMS 定义了五种不同的消息正文格式，以及调用的消息类型，允许你发送并接收一些不同形式的数据，提供现有消息格式的一些级别的兼容性。
· TextMessage--一个字符串对象 *
· MapMessage--一套名称-值对
· ObjectMessage--一个序列化的 Java 对象 *
· BytesMessage--一个字节的数据流 *
· StreamMessage -- Java原始值的数据流
生产者：
@RunWith(SpringJUnit4ClassRunner.class)
@SpringBootTest(classes = ProducerApplication.class)
public class SpringBootProducer {
//JmsMessagingTemplate: 用于工具类发送消息
@Autowired
private JmsMessagingTemplate jmsMessagingTemplate;
@Autowired
private JmsTemplate jmsTemplate;
@Value("${activemq.name}")
private String name;
/**
* 发送TextMessage消息
*/
@Test
public void testMessage(){
jmsTemplate.send(name, new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
TextMessage textMessage = session.createTextMessage("文本消息");
return textMessage;
}
});
}
/**
* 发送MapMessage消息
*/
@Test
public void mapMessage(){
jmsTemplate.send(name, new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
MapMessage mapMessage = session.createMapMessage();
mapMessage.setString("name","张三");
mapMessage.setInt("age",20);
return mapMessage;
}
});
}
/**
* 发送ObjectMessage消息
*//*
@Test
public void objectMessage(){
jmsTemplate.send(name, new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
User user = new User("小明","123456");
ObjectMessage objectMessage = session.createObjectMessage(user);
return objectMessage;
}
});
}
*/
/**
* 发送BytesMessage消息
*/
@Test
public void bytesMessage(){
jmsTemplate.send(name, new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
BytesMessage bytesMessage = session.createBytesMessage();
//1.读取文件
File file = new File("d:/activemq/spring.jpg");
//2.构建文件输入流
try {
FileInputStream inputStream = new FileInputStream(file);
//3.把文件流写入到缓存数组中
byte[] buffer = new byte[(int)file.length()];
inputStream.read(buffer);
//4.把缓存数组写入到BytesMessage中
bytesMessage.writeBytes(buffer);
} catch (Exception e) {
e.printStackTrace();
}
return bytesMessage;
}
});
}
/**
* 发送StreamMessage消息
*/
@Test
public void streamMessage(){
jmsTemplate.send(name, new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
StreamMessage streamMessage = session.createStreamMessage();
streamMessage.writeString("你好，ActiveMQ");
streamMessage.writeInt(20);
//设置消息属性：标记、过滤
streamMessage.setStringProperty("订单","order");
return streamMessage;
}
});
}
}
消费者：
@Component // 放入IOC容器
public class MsgListener {
/**
* 接收TextMessage的方法
*/
/*@JmsListener(destination = "${activemq.name}")
public void receiveMessage(Message message){
if(message instanceof TextMessage){
TextMessage textMessage = (TextMessage)message;
try {
System.out.println("接收消息："+textMessage.getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
}*/
/*@JmsListener(destination = "${activemq.name}")
public void receiveMessage(Message message){
if(message instanceof MapMessage){
MapMessage mapMessage = (MapMessage)message;
try {
System.out.println("名称："+mapMessage.getString("name"));
System.out.println("年龄："+mapMessage.getString("age"));
} catch (JMSException e) {
e.printStackTrace();
}
}
}*/
/* @JmsListener(destination = "${activemq.name}")
public void receiveMessage(Message message){
if(message instanceof ObjectMessage){
ObjectMessage objectMessage = (ObjectMessage)message;
try {
User user = (User)objectMessage.getObject();
System.out.println(user.getUsername());
System.out.println(user.getPassword());
} catch (JMSException e) {
e.printStackTrace();
}
}
}*/
/*@JmsListener(destination = "${activemq.name}")
public void receiveBytesMessage(Message message){
if(message instanceof BytesMessage){
BytesMessage bytesMessage = (BytesMessage)message;
try {
System.out.println("接收消息内容："+bytesMessage.getBodyLength());
//1.设计缓存数组
byte[] buffer = new byte[(int)bytesMessage.getBodyLength()];
//2.把字节消息的内容写入到缓存数组
bytesMessage.readBytes(buffer);
//3.构建文件输出流
FileOutputStream outputStream = new FileOutputStream("d:/activemq/test.jpg");
//4.把数据写出本地硬盘
outputStream.write(buffer);
} catch (Exception e) {
e.printStackTrace();
}
}
}*/
@JmsListener(destination = "${activemq.name}")
public void receiveStreamMessage(Message message){
if(message instanceof StreamMessage){
StreamMessage streamMessage = (StreamMessage)message;
try {
//接收消息属性
System.out.println(streamMessage.getStringProperty("订单"));
System.out.println(streamMessage.readString());
System.out.println(streamMessage.readInt());
} catch (Exception e) {
e.printStackTrace();
}
}
}
}
注意：ActiveMQ5.12后 ，为了安全考虑，ActiveMQ默认不接受自定义的序列化对象，需要将自定义的加入到受信任的列表
# springboot与activemq整合配置
activemq:
broker-url: tcp://192.168.1.144:61616 # 连接地址
user: admin # activemq用户名
password: admin :# activemq密码
packages:
trust-all: true # 让ActiveMQ信任全部自定义对象，实现对象的序列化或反序列化
消息属性
如果需要除消息头字段之外的值，那么可以使用消息属性。他是识别/去重/重点标注等操作，非常有用的方法。
他们是以属性名和属性值对的形式制定的。可以将属性是为消息头得扩展，属性指定一些消息头没有包括的附加信息，比如可以在属性里指定消息选择器。消息的属性就像可以分配给一条消息的附加消息头一样。它们允许开发者添加有关消息的不透明附加信息。它们还用于暴露消息选择器在消息过滤时使用的数据。
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
public class JmsProduce_topic {
public static final String ACTIVEMQ_URL = "tcp://192.168.1.144:61616";
public static final String TOPIC_NAME = "topic01";
public static void main(String[] args) throws Exception{
ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
Connection connection = activeMQConnectionFactory.createConnection();
connection.start();
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Topic topic = session.createTopic(TOPIC_NAME);
MessageProducer messageProducer = session.createProducer(topic);
for (int i = 1; i < 4 ; i++) {
TextMessage textMessage = session.createTextMessage("topic_name--" + i);
// 调用Message的set*Property()方法，就能设置消息属性。根据value的数据类型的不同，有相应的API。
textMessage.setStringProperty("From","123456@qq.com");
textMessage.setByteProperty("Spec", (byte) 1);
textMessage.setBooleanProperty("Invalide",true);
messageProducer.send(textMessage);
}
messageProducer.close();
session.close();
connection.close();
System.out.println(" **** TOPIC_NAME消息发送到MQ完成 ****");
}
}
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
public class JmsConsummer_topic {
public static final String ACTIVEMQ_URL = "tcp://192.168.1.144:61616";
public static final String TOPIC_NAME = "topic01";
public static void main(String[] args) throws Exception{
ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
javax.jms.Connection connection = activeMQConnectionFactory.createConnection();
connection.start();
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Topic topic = session.createTopic(TOPIC_NAME);
MessageConsumer messageConsumer = session.createConsumer(topic);
messageConsumer.setMessageListener( (message) -> {
if (null != message && message instanceof TextMessage){
TextMessage textMessage = (TextMessage)message;
try {
System.out.println("消息体："+textMessage.getText());
System.out.println("消息属性："+textMessage.getStringProperty("From"));
System.out.println("消息属性："+textMessage.getByteProperty("Spec"));
System.out.println("消息属性："+textMessage.getBooleanProperty("Invalide"));
}catch (JMSException e) {
}
}
});
System.in.read();
messageConsumer.close();
session.close();
connection.close();
}
}