activemq的优先级

http://activemq.apache.org/consumer-priority.html consumer 优先级

http://activemq.apache.org/activemq-message-properties.html 消息优先级

1、设置 consumer 的优先级：

queue = new ActiveMQQueue("TEST.QUEUE?consumer.priority=10");
consumer = session.createConsumer(queue);

priority 的取值从0到127。broker 按照 consumer 的优先级给 queue 的 consumers 排序，首先把消息分发给优先级最高的 consumer。一旦该 consumer 的 prefetch buffer 满了，broker 就把消息分发给优先级次高的，prefetch buffer 不满的 consumer。

// org.apache.activemq.broker.region.Queue
// consumer priority 的比较器
private final Comparator<Subscription> orderedCompare = new Comparator<Subscription>() {

    @Override
    public int compare(Subscription s1, Subscription s2) {
        // We want the list sorted in descending order
        // 倒序，即数值大的优先级高
        int val = s2.getConsumerInfo().getPriority() - s1.getConsumerInfo().getPriority();
        if (val == 0 && messageGroupOwners != null) {
            // then ascending order of assigned message groups to favour less loaded consumers
            // Long.compare in jdk7
            long x = s1.getConsumerInfo().getLastDeliveredSequenceId();
            long y = s2.getConsumerInfo().getLastDeliveredSequenceId();
            val = (x < y) ? -1 : ((x == y) ? 0 : 1);
        }
        return val;
    }
};

// 添加 consumer 的时候，会触发排序
// 在 consumers 列表中，靠前的 consumer，先分发消息
private void addToConsumerList(Subscription sub) {
    if (useConsumerPriority) {
        consumers.add(sub);
        Collections.sort(consumers, orderedCompare);
    } else {
        consumers.add(sub);
    }
}

2、设置 message 的优先级需要在 broker 端和 producer 端配置：

2.1 在 broker 端设置 TEST.BAT 队列为 prioritizedMessages = “true”

<policyEntry queue="TEST.BAT" prioritizedMessages="true" producerFlowControl="true" memoryLimit="1mb">
    <deadLetterStrategy>
        <individualDeadLetterStrategy queuePrefix="TEST"/>
    </deadLetterStrategy>
    <pendingQueuePolicy>
        <storeCursor/>
    </pendingQueuePolicy>
</policyEntry>

2.2 producer 发送消息时，设置 message 的优先级

TextMessage message = session.createTextMessage(text);
producer.send(destination, message, DeliveryMode.NON_PERSISTENT, 1, 0);

设置 message 的优先级，需要调用：

void javax.jms.MessageProducer.send(Destination destination, Message message, int deliveryMode, int priority, long timeToLive)
throws JMSException

而不能这么写：

TextMessage message = session.createTextMessage(text);
message.setJMSPriority(0);

初步看是 ActiveMQ 的 bug。消息的 priority 值，从0到9。消息配置了优先级之后，消息存放在 PrioritizedPendingList 中。

// 省略部分代码
private class PrioritizedPendingListIterator implements Iterator<MessageReference> {
    private int index = 0;
    private int currentIndex = 0;
    List<PendingNode> list = new ArrayList<PendingNode>(size());

    PrioritizedPendingListIterator() {
        for (int i = MAX_PRIORITY - 1; i >= 0; i--) {
            // priority 值大的优先级高
            OrderedPendingList orderedPendingList = lists[i];
            if (!orderedPendingList.isEmpty()) {
                list.addAll(orderedPendingList.getAsList());
            }
        }
    }
}

---------

ActiveMq之消息的优先级-yellowcong


2017年12月09日 12:28:12
阅读数：1053





消息发送后，有的时候收发消息的顺序不是我们发送消息前的优先级发送的，我们可以通过配置activemq.xml,设定哪个队列按照优先级别来进行数据的发送。
修改配置文件
修改配置文件，设定需要配置优先级操作的消息队列
vim  apache-activemq-5.11.1/conf/activemq.xml

#大概44行左右
#设定queue="queue1" 设定 queue1 的队列是按优先级发送的
<policyEntry queue="queue1" prioritizedMessages="true" />

#重启服务生效
apache-activemq-5.11.1/bin/activemq-admin start


测试案例
package com.yellowcong.provice;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

/**
 * 创建用户:狂飙的yellowcong<br/>
 * 创建日期:2017年12月9日<br/>
 * 创建时间:上午11:01:51<br/>
 * 机能概要:
 */
public class DemoSendMessage {

    // activemq的服务器地址
    private static final String ACTIVEMQ_HOST = "tcp://192.168.100.10:61616";

    // 用户名
    private static final String USERNAME = "yellowcong";
    // 密码
    private static final String PASSWORD = "yellowcong";

    public static void main(String[] args) throws Exception {
        provider();

        customer();
    }

    /**
     * 创建用户:狂飙的yellowcong<br/>
     * 创建日期:2017年12月9日<br/>
     * 创建时间:上午10:52:05<br/>
     * 机能概要:消费者确认消息
     * 
     * @throws Exception
     */
    public static void customer() throws Exception {
        // 获取连接
        Connection conn = null;

        try {
            conn = getConnection();
            // 获取session
            Session session = conn.createSession(true, Session.CLIENT_ACKNOWLEDGE);
            // 需要和activemq.xml配置文件的队列对上
            Destination destination = session.createQueue("queue1");

            // 获取消费者
            MessageConsumer cus = session.createConsumer(destination);
            System.out.println("-------------接收并确认消息----------------");
            while (true) {
                // 接收消息
                TextMessage msg = (TextMessage) cus.receive();
                if (msg == null) {
                    break;
                }
                // 确认接收， 又开启一个线程，去发送给服务器，按收到消息了
                msg.acknowledge();
                System.out.println(msg.getText());

            }

        } finally {
            // 关闭连接
            conn.close();
        }
    }

    /**
     * 创建用户:狂飙的yellowcong<br/>
     * 创建日期:2017年12月9日<br/>
     * 创建时间:上午10:48:03<br/>
     * 机能概要: 生产消息
     * 
     * @throws Exception
     */
    public static void provider() throws Exception {
        // 获取连接
        Connection conn = null;
        try {
            conn = getConnection();
            // 获取session
            Session session = conn.createSession(true, Session.CLIENT_ACKNOWLEDGE);
            // 创建队列
            // 需要和activemq.xml配置文件的队列对上
            Destination destination = session.createQueue("queue1");
            // 获取生产者
            MessageProducer pro = session.createProducer(destination);

            // 创建消息
            TextMessage msg = session.createTextMessage();

            // 发送消息
            // 第一个参数 消息的目的地 Destination
            // 第二个参数 发送的消息
            // 第三个参数 消息的模式 DeliveryMode
            // 第四个参数 消息的优先级
            // 第五个参数 消息存活时间 存活时间单位是ms
            for (int i = 0; i < 10; i++) {
                msg.setText("请确认消息,消息添加顺序"+i+",优先级" + i);
                pro.send(destination, msg, DeliveryMode.NON_PERSISTENT, i, 1000 * 10);
            }

            // 提交事物
            session.commit();

            System.out.println("-----------------发送消息----------------");
        } finally {
            // 关闭连接
            conn.close();
        }

    }

    /**
     * 创建用户:狂飙的yellowcong<br/>
     * 创建日期:2017年12月9日<br/>
     * 创建时间:上午10:42:48<br/>
     * 机能概要: 获取连接
     * 
     * @return
     * @throws Exception
     */
    public static Connection getConnection() throws Exception {
        // 1.获取工厂连接类
        ConnectionFactory fc = new ActiveMQConnectionFactory(USERNAME, PASSWORD, ACTIVEMQ_HOST);

        // 2.获取连接
        Connection conn = fc.createConnection();
        conn.start();

        System.out.println("-----------------获取连接----------------");

        return conn;

    }
}

运行结果
搭建可以发现，优先级别高的先消费。