[ https://issues.apache.org/jira/browse/ROCKETMQ-138?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16175808#comment-16175808 ]
ASF GitHub Bot commented on ROCKETMQ-138: ----------------------------------------- Github user dongeforever commented on a diff in the pull request: https://github.com/apache/incubator-rocketmq/pull/138#discussion_r140401292 --- Diff: client/src/main/java/org/apache/rocketmq/client/producer/selector/SelectMessageQueueByConsistentHash.java --- @@ -0,0 +1,132 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.rocketmq.client.producer.selector; + +import org.apache.rocketmq.client.producer.MessageQueueSelector; +import org.apache.rocketmq.common.message.Message; +import org.apache.rocketmq.common.message.MessageQueue; + +import java.util.HashMap; +import java.util.List; +import java.util.SortedMap; +import java.util.TreeMap; + +/** +* wanting to send a message with the same key combine with orderly consumer, user can +* use the implements of the MessageQueueSelector, which has one optimization compared with SelectMessageQueueByHash +* when a broker crash , it can reduce the message Arriving out of the order +*/ +public class SelectMessageQueueByConsistentHash implements MessageQueueSelector { + + private volatile SortedMap<Integer, String> virtualNodes = + new TreeMap<Integer, String>(); + + private static final int DEFAULT_VIRTUAL_NODES = 100; + + private final int virtualNodeNum; + + private volatile HashMap<String, MessageQueue> idToQueueMap = new HashMap<String, MessageQueue>(); + + public SelectMessageQueueByConsistentHash() { + this.virtualNodeNum = DEFAULT_VIRTUAL_NODES; + } + + public SelectMessageQueueByConsistentHash(int virtualNodeNum) { + this.virtualNodeNum = virtualNodeNum; + } + + @Override + public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) { + synchronized (this) { + if (queueChange(mqs)) { + reloadConsistentHash(mqs); + } + + String uniqueQueueId = getMsgQueueIdBy(arg.toString()); + MessageQueue messageQueue = idToQueueMap.get(uniqueQueueId); + return messageQueue; + } + } + + private boolean queueChange(List<MessageQueue> mqs) { + if (mqs.size() != this.idToQueueMap.size()) { + return true; + } + + for (MessageQueue queue : mqs) { + String id = queue.getTopic() + "_" + queue.getBrokerName() + "_" + queue.getQueueId(); + if (!this.idToQueueMap.containsKey(id)) { + return true; + } + } + + return false; + } + + private String getMsgQueueIdBy(String arg) { + int hash = getHash(arg); + SortedMap<Integer, String> subMap = virtualNodes.tailMap(hash); + + Integer i; + String virtualNode; + + if (subMap.size() == 0) { + i = virtualNodes.firstKey(); + virtualNode = virtualNodes.get(i); + } else { + i = subMap.firstKey(); + virtualNode = subMap.get(i); + } + + String result = virtualNode.substring(0, virtualNode.indexOf("&&")); + return result; + } + + private int getHash(String str) { --- End diff -- Why not just use str.hashCode? > Add AuthenticationException class to remove hard coded Aliyun authentication > class > ---------------------------------------------------------------------------------- > > Key: ROCKETMQ-138 > URL: https://issues.apache.org/jira/browse/ROCKETMQ-138 > Project: Apache RocketMQ > Issue Type: Improvement > Components: rocketmq-remoting > Reporter: Jaskey Lam > Assignee: Jaskey Lam > Priority: Minor > Fix For: 4.1.0-incubating > > > in NettyRemotingAbstract.java > a hard coded aliyun class is used > {code} > catch (Throwable e) { > if > (!"com.aliyun.openservices.ons.api.impl.authority.exception.AuthenticationException" > .equals(e.getClass().getCanonicalName())) { > PLOG.error("process request exception", e); > PLOG.error(cmd.toString()); > } > {code} > A common AuthenticationException should be added to identify Authentication > failure. Developers can throw this exception so that remoting component can > ignore it -- This message was sent by Atlassian JIRA (v6.4.14#64029)