聊聊rocketmq的MessageQueueSelector

序

本文主要研究一下 rocketmq 的 MessageQueueSelector

MessageQueueSelector

rocketmq-client-4.5.2-sources.jar!/org/apache/rocketmq/client/producer/MessageQueueSelector.java

public interface MessageQueueSelector {MessageQueue select(final List<MessageQueue> mqs, final Message msg, final Object arg);
}

• MessageQueueSelector 接口定义了 select 方法，返回 MessageQueue；它有几个实现类，分别是 SelectMessageQueueByHash、SelectMessageQueueByRandom、SelectMessageQueueByMachineRoom

SelectMessageQueueByHash

rocketmq-client-4.5.2-sources.jar!/org/apache/rocketmq/client/producer/selector/SelectMessageQueueByHash.java

public class SelectMessageQueueByHash implements MessageQueueSelector {

    @Override
    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {int value = arg.hashCode();
        if (value < 0) {value = Math.abs(value);
        }

        value = value % mqs.size();
        return mqs.get(value);
    }
}

• SelectMessageQueueByHash 实现了 MessageQueueSelector 接口，其 select 方法取 arg 参数的 hashcode 的绝对值，然后对 mqs.size() 取余，得到目标队列在 mqs 的下标

SelectMessageQueueByRandom

rocketmq-client-4.5.2-sources.jar!/org/apache/rocketmq/client/producer/selector/SelectMessageQueueByRandom.java

public class SelectMessageQueueByRandom implements MessageQueueSelector {private Random random = new Random(System.currentTimeMillis());

    @Override
    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {int value = random.nextInt(mqs.size());
        return mqs.get(value);
    }
}

• SelectMessageQueueByRandom 实现了 MessageQueueSelector 接口，其 select 方法直接根据 mqs.size() 随机一个值作为目标队列在 mqs 的下标

SelectMessageQueueByMachineRoom

rocketmq-client-4.5.2-sources.jar!/org/apache/rocketmq/client/producer/selector/SelectMessageQueueByMachineRoom.java

public class SelectMessageQueueByMachineRoom implements MessageQueueSelector {
    private Set<String> consumeridcs;

    @Override
    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {return null;}

    public Set<String> getConsumeridcs() {return consumeridcs;}

    public void setConsumeridcs(Set<String> consumeridcs) {this.consumeridcs = consumeridcs;}
}

• SelectMessageQueueByMachineRoom 实现了 MessageQueueSelector 接口，其 select 方法目前返回 null

RocketMQTemplate

rocketmq-spring-boot-2.0.3-sources.jar!/org/apache/rocketmq/spring/core/RocketMQTemplate.java

public class RocketMQTemplate extends AbstractMessageSendingTemplate<String> implements InitializingBean, DisposableBean {private static final  Logger log = LoggerFactory.getLogger(RocketMQTemplate.class);

    private DefaultMQProducer producer;

    private ObjectMapper objectMapper;

    private String charset = "UTF-8";

    private MessageQueueSelector messageQueueSelector = new SelectMessageQueueByHash();

    private final Map<String, TransactionMQProducer> cache = new ConcurrentHashMap<>(); //only put TransactionMQProducer by now!!!

    public DefaultMQProducer getProducer() {return producer;}

    public void setProducer(DefaultMQProducer producer) {this.producer = producer;}

    public ObjectMapper getObjectMapper() {return objectMapper;}

    public void setObjectMapper(ObjectMapper objectMapper) {this.objectMapper = objectMapper;}

    public String getCharset() {return charset;}

    public void setCharset(String charset) {this.charset = charset;}

    public MessageQueueSelector getMessageQueueSelector() {return messageQueueSelector;}

    public void setMessageQueueSelector(MessageQueueSelector messageQueueSelector) {this.messageQueueSelector = messageQueueSelector;}

    //......
}

• RocketMQTemplate 默认创建的 MessageQueueSelector 是 SelectMessageQueueByHash

小结

MessageQueueSelector 接口定义了 select 方法，返回 MessageQueue；它有几个实现类，分别是 SelectMessageQueueByHash、SelectMessageQueueByRandom、SelectMessageQueueByMachineRoom；RocketMQTemplate 默认创建的 MessageQueueSelector 是 SelectMessageQueueByHash

doc

• MessageQueueSelector