ActiveMQ学习笔记-消息特性

柔光的暖阳◎ 2022-06-16 14:11 348阅读 0赞

## ActiveMQ消息属性 ##

### 消息属性 ###

<table> 
 <thead> 
  <tr> 
   <th>属性名</th> 
   <th>类型</th> 
   <th>默认值</th> 
   <th>描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>JMSDestination</td> 
   <td>javax.jms.Destination</td> 
   <td>生产者set进去</td> 
   <td>发送消息目的地</td> 
  </tr> 
  <tr> 
   <td>JMSReplyTo</td> 
   <td>javax.jms.Destination</td> 
   <td>null</td> 
   <td>用户定义</td> 
  </tr> 
  <tr> 
   <td>JMSType</td> 
   <td>String</td> 
   <td>“”</td> 
   <td>用户定义</td> 
  </tr> 
  <tr> 
   <td>JMSDeliveryMode</td> 
   <td>int</td> 
   <td>DeliveryMode.PERSISTENT</td> 
   <td>消息是否要持久化</td> 
  </tr> 
  <tr> 
   <td>JMSPriority</td> 
   <td>int</td> 
   <td>4</td> 
   <td>0-9</td> 
  </tr> 
  <tr> 
   <td>JMSMessageID</td> 
   <td>String</td> 
   <td>unique</td> 
   <td>消息唯一标示符</td> 
  </tr> 
  <tr> 
   <td>JMSTimestamp</td> 
   <td>long</td> 
   <td>消息发送时间</td> 
   <td>毫秒</td> 
  </tr> 
  <tr> 
   <td>JMSCorrelationID</td> 
   <td>String</td> 
   <td>null</td> 
   <td>用户定义</td> 
  </tr> 
  <tr> 
   <td>JMSExpiration</td> 
   <td>long</td> 
   <td>0</td> 
   <td>0表示用不过气，消息过期时间</td> 
  </tr> 
  <tr> 
   <td>JMSRedelivered</td> 
   <td>boolean</td> 
   <td>false</td> 
   <td>重新传递</td> 
  </tr> 
 </tbody> 
</table>

### JMS定义属性 ###

<table> 
 <thead> 
  <tr> 
   <th>属性名</th> 
   <th>类型</th> 
   <th>默认值</th> 
   <th>描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>JMSXDeliveryCount</td> 
   <td>int</td> 
   <td>0</td> 
   <td>尝试发送消息的次数</td> 
  </tr> 
  <tr> 
   <td>JMSXGroupID</td> 
   <td>String</td> 
   <td>null</td> 
   <td>表示消息的分组</td> 
  </tr> 
  <tr> 
   <td>JMSXGroupSeq</td> 
   <td>int</td> 
   <td>0</td> 
   <td>消息的序列号</td> 
  </tr> 
  <tr> 
   <td>JMSXProducerTXID</td> 
   <td>String</td> 
   <td>null</td> 
   <td>事务标示</td> 
  </tr> 
 </tbody> 
</table>

### ActiveMQ定义属性 ###

<table> 
 <thead> 
  <tr> 
   <th>属性名</th> 
   <th>类型</th> 
   <th>默认值</th> 
   <th>描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>JMSActiveMQBrokerInTime</td> 
   <td>long</td> 
   <td>0</td> 
   <td>消息到达broker的时间</td> 
  </tr> 
  <tr> 
   <td>JMSActiveMQBrokerOutTime</td> 
   <td>long</td> 
   <td>0</td> 
   <td>消息离开broker的时间</td> 
  </tr> 
 </tbody> 
</table>

## 公告消息（Advisory Message） ##

ActiveMQ支持的公告消息（相当于监听器）

*  消费者，生产者和connection开始和停止
 *  临时目的地的创建和销毁
 *  话题和队列上的消息过期
 *  broker发送消息到没有消费者的目的地
 *  connections开启和停止
    
    公告消息是一种管理JMS的渠道，我们可以接收到在broker上的生产者消费者目的地的变化。  
    当我们通过JMX查看broker时候，可以看到公告话题以ActiveMQ.Advisory为前缀，每个公告拥有类型为Advisory的消息和一些预定义的消息属性

<table> 
 <thead> 
  <tr> 
   <th>属性名</th> 
   <th>类型</th> 
   <th>描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>originBrokerId</td> 
   <td>StringProperty</td> 
   <td>公告起源的broker的ID</td> 
  </tr> 
  <tr> 
   <td>originBrokerName</td> 
   <td>StringProperty</td> 
   <td>公告起源的broker的名称</td> 
  </tr> 
  <tr> 
   <td>originBrokerURL</td> 
   <td>StringProperty</td> 
   <td>公告起源的第一个broker url</td> 
  </tr> 
 </tbody> 
</table>

我们可以获得到消息的一些元数据

...
    
        Destination advisoryDestination = AdvisorySupport.getProducerAdvisoryTopic(destination)
        MessageConsumer consumer = session.createConsumer(advisoryDestination);
        consumer.setMessageListener(this);
    ....
    public void onMessage(Message msg){
        if (msg instanceof ActiveMQMessage){
            try {
                 ActiveMQMessage aMsg =  (ActiveMQMessage)msg;
                 ProducerInfo prod = (ProducerInfo) aMsg.getDataStructure();
            } catch (JMSException e) {
                log.error("Failed to process message: " + msg);
            }
        }
    }

### 支持的公告话题 ###

**客户端的公告消息**  
我们可以用consumerCount头信息获取当前消费者的数量来得知当前的公告的消息是否已经发送。

<table> 
 <thead> 
  <tr> 
   <th>Advisory Topics</th> 
   <th>Description</th> 
   <th>properties</th> 
   <th>Data Structure</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>ActiveMQ.Advisory.Connection</td> 
   <td>ActiveMQ.Advisory.Connection</td> 
   <td></td> 
   <td>ConnectionInfo, RemoveInfo</td> 
  </tr> 
  <tr> 
   <td>ActiveMQ.Advisory.Producer.Queue</td> 
   <td>Producer start &amp; stop messages on a Queue</td> 
   <td>String=’producerCount’ - the number of producers</td> 
   <td>ProducerInfo</td> 
  </tr> 
  <tr> 
   <td>ActiveMQ.Advisory.Producer.Topic</td> 
   <td>Producer start &amp; stop messages on a Topic</td> 
   <td>String=’producerCount’ - the number of producers</td> 
   <td>ProducerInfo</td> 
  </tr> 
  <tr> 
   <td>ActiveMQ.Advisory.Consumer.Queue</td> 
   <td>Consumer start &amp; stop messages on a Queue</td> 
   <td>String=’consumerCount’ - the number of Consumers</td> 
   <td>ConsumerInfo, RemoveInfo</td> 
  </tr> 
  <tr> 
   <td>ActiveMQ.Advisory.Consumer.Topic</td> 
   <td>Consumer start &amp; stop messages on a Topic</td> 
   <td>String=’consumerCount’ - the number of Consumers</td> 
   <td>ConsumerInfo, RemoveInfo</td> 
  </tr> 
 </tbody> 
</table>

**目的地和消息的公告**

<table> 
 <thead> 
  <tr> 
   <th>Advisory Topics</th> 
   <th>描述</th> 
   <th>属性</th> 
   <th>数据结构</th> 
   <th>默认值</th> 
   <th>策略属性</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>ActiveMQ.Advisory.Queue</td> 
   <td>Queue create &amp; destroy</td> 
   <td>null</td> 
   <td>DestinationInfo</td> 
   <td>true</td> 
   <td>none</td> 
  </tr> 
  <tr> 
   <td>ActiveMQ.Advisory.Topic</td> 
   <td>Topic create &amp; destroy</td> 
   <td>null</td> 
   <td>DestinationInfo</td> 
   <td>true</td> 
   <td>none</td> 
  </tr> 
  <tr> 
   <td>ActiveMQ.Advisory.TempQueue</td> 
   <td>Temporary Queue create &amp; destroy</td> 
   <td>null</td> 
   <td>DestinationInfo</td> 
   <td>true</td> 
   <td>none</td> 
  </tr> 
  <tr> 
   <td>ActiveMQ.Advisory.TempTopic</td> 
   <td>Temporary Topic create &amp; destroy</td> 
   <td>null</td> 
   <td>DestinationInfo</td> 
   <td>true</td> 
   <td>none</td> 
  </tr> 
  <tr> 
   <td>ActiveMQ.Advisory.Expired.Queue</td> 
   <td>Expired messages on a Queue</td> 
   <td>String=’orignalMessageId’ - the expired id</td> 
   <td>Message</td> 
   <td>true</td> 
   <td>none</td> 
  </tr> 
  <tr> 
   <td>ActiveMQ.Advisory.Expired.Topic</td> 
   <td>Expired messages on a Topic</td> 
   <td>String=’orignalMessageId’ - the expired id</td> 
   <td>Message</td> 
   <td>true</td> 
   <td>none</td> 
  </tr> 
  <tr> 
   <td>ActiveMQ.Advisory.NoConsumer.Queue</td> 
   <td>No consumer is available to process messages being sent on a Queue</td> 
   <td>null</td> 
   <td>Message</td> 
   <td>false</td> 
   <td>sendAdvisoryIfNoConsumers</td> 
  </tr> 
  <tr> 
   <td>ActiveMQ.Advisory.NoConsumer.Topic</td> 
   <td>No consumer is available to process messages being sent on a Topic</td> 
   <td>null</td> 
   <td>Message</td> 
   <td>false</td> 
   <td>sendAdvisoryIfNoConsumers</td> 
  </tr> 
 </tbody> 
</table>

ActiveMQ5.2新的公告消息

<table> 
 <thead> 
  <tr> 
   <th>Advisory Topics</th> 
   <th>描述</th> 
   <th>属性</th> 
   <th>数据结构</th> 
   <th>默认值</th> 
   <th>策略属性</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>ActiveMQ.Advisory.SlowConsumer.Queue</td> 
   <td>Slow Queue Consumer</td> 
   <td>String=’consumerId’ - the consumer id</td> 
   <td>ConsumerInfo</td> 
   <td>false</td> 
   <td>advisoryForSlowConsumers</td> 
  </tr> 
  <tr> 
   <td>ActiveMQ.Advisory.SlowConsumer.Topic</td> 
   <td>Slow Topic Consumer</td> 
   <td>String=’consumerId’ - the consumer id</td> 
   <td>ConsumerInfo</td> 
   <td>false</td> 
   <td>advisoryForSlowConsumers</td> 
  </tr> 
  <tr> 
   <td>ActiveMQ.Advisory.FastProducer.Queue</td> 
   <td>Fast Queue producer</td> 
   <td>String=’producerId’ - the producer id</td> 
   <td>ProducerInfo</td> 
   <td>false</td> 
   <td>advisdoryForFastProducers</td> 
  </tr> 
  <tr> 
   <td>ActiveMQ.Advisory. FastProducer.Topic</td> 
   <td>Fast Topic producer</td> 
   <td>String=’consumerId’ - the producer id</td> 
   <td>ProducerInfo</td> 
   <td>false</td> 
   <td>advisdoryForFastProducers</td> 
  </tr> 
  <tr> 
   <td>ActiveMQ.Advisory.MessageDiscarded.Queue</td> 
   <td>Message discarded</td> 
   <td>String=’orignalMessageId’ - the discarded id</td> 
   <td>Message</td> 
   <td>false</td> 
   <td>advisoryForDiscardingMessages</td> 
  </tr> 
  <tr> 
   <td>ActiveMQ.Advisory.MessageDiscarded.Topic</td> 
   <td>Message discarded</td> 
   <td>String=’orignalMessageId’ - the discarded id</td> 
   <td>Message</td> 
   <td>false</td> 
   <td>advisoryForDiscardingMessages</td> 
  </tr> 
  <tr> 
   <td>ActiveMQ.Advisory.MessageDelivered.Queue</td> 
   <td>Message delivered to the broker</td> 
   <td>String=’orignalMessageId’ - the delivered id</td> 
   <td>Message</td> 
   <td>false</td> 
   <td>advisoryForDelivery</td> 
  </tr> 
  <tr> 
   <td>ActiveMQ.Advisory.MessageDelivered.Topic</td> 
   <td>Message delivered to the broker</td> 
   <td>String=’orignalMessageId’ - the delivered id</td> 
   <td>Message</td> 
   <td>false</td> 
   <td>advisoryForDelivery</td> 
  </tr> 
  <tr> 
   <td>ActiveMQ.Advisory.MessageConsumed.Queue</td> 
   <td>Message consumed by a client</td> 
   <td>String=’orignalMessageId’ - the delivered id</td> 
   <td>Message</td> 
   <td>false</td> 
   <td>advisoryForConsumed</td> 
  </tr> 
  <tr> 
   <td>ActiveMQ.Advisory.MessageConsumed.Topic</td> 
   <td>Message consumed by a client</td> 
   <td>String=’orignalMessageId’ - the delivered id</td> 
   <td>Message</td> 
   <td>false</td> 
   <td>advisoryForConsumed</td> 
  </tr> 
  <tr> 
   <td>ActiveMQ.Advisory.FULL</td> 
   <td>A Usage resource is at its limit</td> 
   <td>String=’usageName’ - the name of Usage resource</td> 
   <td>null</td> 
   <td>false</td> 
   <td>advisoryWhenFull</td> 
  </tr> 
  <tr> 
   <td>ActiveMQ.Advisory.MasterBroker</td> 
   <td>A broker is now the master in a master/slave configuration</td> 
   <td>null</td> 
   <td>null</td> 
   <td>true</td> 
   <td>none</td> 
  </tr> 
 </tbody> 
</table>

### 5.4后新公告 ###

<table> 
 <thead> 
  <tr> 
   <th>Advisory Topics</th> 
   <th>描述</th> 
   <th>属性</th> 
   <th>数据结构</th> 
   <th>默认值</th> 
   <th>策略属性</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>ActiveMQ.Advisory.MessageDLQd.Queue</td> 
   <td>Message sent to DLQ</td> 
   <td>String=’orignalMessageId’ - the delivered id</td> 
   <td>Message</td> 
   <td>Always on</td> 
   <td>advisoryForConsumed</td> 
  </tr> 
  <tr> 
   <td>ActiveMQ.Advisory.MessageDLQd.Topic</td> 
   <td>Message sent to DLQ</td> 
   <td>String=’orignalMessageId’ - the delivered id</td> 
   <td>Message</td> 
   <td>Always on</td> 
   <td>advisoryForConsumed</td> 
  </tr> 
 </tbody> 
</table>

### Network bridge advisories ###

<table> 
 <thead> 
  <tr> 
   <th>Advisory Topics</th> 
   <th>描述</th> 
   <th>属性</th> 
   <th>数据结构</th> 
   <th>默认值</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>ActiveMQ.Advisory.NetworkBridge</td> 
   <td>Network bridge being stopped or started</td> 
   <td>Boolean=”started” - true if bridge is started, false if it is stopped Boolean=”createdByDuplex” - true if the bridge is created by remote network connector</td> 
   <td>BrokerInfo - provides data of the remote broker</td> 
   <td>Always on</td> 
  </tr> 
 </tbody> 
</table>

### 开启公告消息 ###

公告消息默认是不开启的，我们可以使用下面的配置开启公告配置

<destinationPolicy>
       <policyMap><policyEntries> 
          <policyEntry topic=">" advisoryForConsumed="true" />
       </policyEntries></policyMap>
    </destinationPolicy>

### 关闭公告消息 ###

xml配置如下

<broker advisorySupport="false">...

java code 配置

BrokerService broker = new BrokerService();
    broker.setAdvisorySupport(false);
    ...
    broker.start();

url链接串

"tcp://localhost:61616?jms.watchTopicAdvisories=false"

设置ConnectionFactory的属性

ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory();
    factory.setWatchTopicAdvisories(false);

### ActiveMQ一些静态方法 ###

AdvisorySupport.getConsumerAdvisoryTopic()
    AdvisorySupport.getProducerAdvisoryTopic()
    AdvisorySupport.getExpiredTopicMessageAdvisoryTopic()
    AdvisorySupport.getExpiredQueueMessageAdvisoryTopic()
    AdvisorySupport.getNoTopicConsumersAdvisoryTopic()
    AdvisorySupport.getNoQueueConsumersAdvisoryTopic()
    AdvisorySupport.getDestinationAdvisoryTopic()
    AdvisorySupport.getExpiredQueueMessageAdvisoryTopic()
    AdvisorySupport.getExpiredTopicMessageAdvisoryTopic()
    AdvisorySupport.getNoQueueConsumersAdvisoryTopic()
    AdvisorySupport.getNoTopicConsumersAdvisoryTopic()
    
    //Version 5.2 onwards
    
    
    AdvisorySupport.getSlowConsumerAdvisoryTopic()
    AdvisorySupport.getFastProducerAdvisoryTopic()
    AdvisorySupport.getMessageDiscardedAdvisoryTopic()
    AdvisorySupport.getMessageDeliveredAdvisoryTopic()
    AdvisorySupport.getMessageConsumedAdvisoryTopic()
    AdvisorySupport.getMasterBrokerAdvisoryTopic()
    AdvisorySupport.getFullAdvisoryTopic()

## Blob 消息（Blob Message） ##

发送大文件信息  
文件地址，或者url等共享文件系统的文档

BlobMessage message = session.createBlobMessage(new URL("http://some.shared.site.com");
    producer.send(message);

// lets use a local file
    BlobMessage message = session.createBlobMessage(new File("/foo/bar");
    producer.send(message);

// lets use a stream
    InputStream in = ...;
    BlobMessage message = session.createBlobMessage(in);
    producer.send(message);

### 接收信息 ###

public class MyListener implements MessageListener {
        
      public void onMessage(Message message) {
        if (message instanceof BlobMessage) {
          BlobMessage blobMessage = (BlobMessage) message;
          InputStream in = blobMessage.getInputStream();
    
          // process the stream...
        }
      }
    }

## Delay and Schedule Message Delivery ##

### 开启定时发送配置 ###

定时发送的策略org.apache.activemq.ScheduledMessage的属性名称

<table> 
 <thead> 
  <tr> 
   <th>属性名称</th> 
   <th>类型</th> 
   <th>描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>AMQ_SCHEDULED_DELAY</td> 
   <td>long</td> 
   <td>毫秒单位，等待多久时间发送到broker</td> 
  </tr> 
  <tr> 
   <td>AMQ_SCHEDULED_PERIOD</td> 
   <td>long</td> 
   <td>第一次发送后等待调度的周期调度的时间</td> 
  </tr> 
  <tr> 
   <td>AMQ_SCHEDULED_REPEAT</td> 
   <td>int</td> 
   <td>重复调度的次数</td> 
  </tr> 
  <tr> 
   <td>AMQ_SCHEDULED_CRON</td> 
   <td>String</td> 
   <td>cron的配置策略</td> 
  </tr> 
 </tbody> 
</table>

第一次等待发送60秒

MessageProducer producer = session.createProducer(destination);
    TextMessage message = session.createTextMessage("test msg");
    long time = 60 * 1000;
    message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, time);
    producer.send(message);

第一次发送延时30，重复10次，每次间隔10秒重复发送

MessageProducer producer = session.createProducer(destination);
    TextMessage message = session.createTextMessage("test msg");
    long delay = 30 * 1000;
    long period = 10 * 1000;
    int repeat = 9;
    message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, delay);
    message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_PERIOD, period);
    message.setIntProperty(ScheduledMessage.AMQ_SCHEDULED_REPEAT, repeat);
    producer.send(message);

使用cron的形式配置

cron的配置是优先级于其他的定时配置的，  
那么这样子的配置的话就是每小时发送一次，第一次发送延迟1秒，发送10次，间隔1秒

MessageProducer producer = session.createProducer(destination);
    TextMessage message = session.createTextMessage("test msg");
    message.setStringProperty(ScheduledMessage.AMQ_SCHEDULED_CRON, "0 * * * *");
    message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, 1000);
    message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_PERIOD, 1000);
    message.setIntProperty(ScheduledMessage.AMQ_SCHEDULED_REPEAT, 9);
    producer.send(message);

## JMS流（JMS Streams） ##

该模块已经遗弃，不推荐使用

## 消息转换(Message Transformation) ##

有时候我们希望对消息进行转换，比如将对象类型的转换为Xml的格式。  
ActiveMQ4.2以后提出了可插拔的策略MessageTransformer接口，该接口允许我们

*  在JMS提供者发送到消息总线前加强或者转换消息
 *  在接收到消息后但是还未发送到消费者前加强或者转换消息

可以使用setTransformer方法的类

*  ActiveMQConnectionFactory
 *  ActiveMQConnection
 *  ActiveMQSession
 *  ActiveMQMessageConsumer
 *  ActiveMQMessageProducer

## 对象消息 ##

ObjectMessage依赖于java语言的序列化，这样子很容易被攻击者攻击，所以在脚本里面会配置白名单$\{ACTIVEMQ\_HOME\}/bin/env脚本。ObjectMessage的使用从5.12.2开始版本是不推荐，但是这里我们还是介绍一下。

-Dorg.apache.activemq.SERIALIZABLE_PACKAGES=java.lang,javax.security,java.util,org.apache.activemq,org.fusesource.hawtbuf,com.thoughtworks.xstream.mapper,com.mycompany.myapp

或者使用通配

-Dorg.apache.activemq.SERIALIZABLE_PACKAGES=*

客户端代码或者Spring配置文件的配置

ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");
    factory.setTrustedPackages(new ArrayList(Arrays.asList("org.apache.activemq.test,org.apache.camel.test".split(","))));

ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");
    factory.setTrustAllPackages(true);

<bean id="connectionFactory" class="org.apache.activemq.spring.ActiveMQConnectionFactory">
        <property name="brokerURL" value="tcp://localhost:61616"/>
        <property name="trustedPackages">
            <list>
                <value>org.apache.activemq.test</value>
                <value>org.apache.camel.test</value>
            </list>
        </property>
    </bean>
    <bean id="jmsConfig" class="org.apache.camel.component.jms.JmsConfiguration">
        <property name="connectionFactory" ref="connectionFactory"/>
    </bean>
    <bean id="activemq" class="org.apache.activemq.camel.component.ActiveMQComponent">
        <property name="configuration" ref="jmsConfig"/>
    </bean>

## 结构化消息属性和Map消息 ##

JMS允许我们使用Map或者List来属性来映射任意语言的配置，这样子消息就可以方便在C，C++等语言转换了。  
\- Message.setObjectProperty(key, value)  
\- MapMessage.setObject(key, value)