package com.alibaba.rocketmq.client.producer.selector;

import com.alibaba.rocketmq.client.producer.MessageQueueSelector;
import com.alibaba.rocketmq.common.message.Message;
import com.alibaba.rocketmq.common.message.MessageQueue;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;

public class SelectMessageQueueByRegion implements MessageQueueSelector {

    private Region target;

    private final AtomicReference<List<MessageQueue>> targetMessageQueues = new AtomicReference<>();

    private final AtomicReference<List<MessageQueue>> messageQueues = new AtomicReference<>();

    private final AtomicLong cursor = new AtomicLong();

    public SelectMessageQueueByRegion(Region target) {
        if (null == target) {
            this.target = Region.SAME;
        } else {
            this.target = target;
        }

        if (Region.SAME == this.target) {
            this.target = Region.parse(Integer.parseInt(Util.LOCAL_DATA_CENTER_ID));
        }

        schedule();
    }

    private void schedule() {
        ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {
            @Override
            public Thread newThread(Runnable runnable) {
                Thread thread = new Thread(runnable, "SelectMessageQueueByRegion");
                thread.setDaemon(true);
                return thread;
            }
        });

        scheduler.scheduleWithFixedDelay(new Runnable() {
            @Override
            public void run() {
                update();
            }
        }, 30, 30, TimeUnit.SECONDS);
    }

    private void update() {
        if (Region.ANY == this.target) {
            targetMessageQueues.set(messageQueues.get());
            return;
        }

        List<MessageQueue> all = messageQueues.get();
        List<MessageQueue> result = new ArrayList<>();
        for (MessageQueue messageQueue : all) {
            if (Util.getRegionIndex(messageQueue) == target.getIndex()) {
                result.add(messageQueue);
            }
        }

        if (!result.isEmpty()) {
            targetMessageQueues.set(result);
        } else {
            targetMessageQueues.set(all);
        }
    }

    @Override
    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
        messageQueues.set(mqs);

        if (null == targetMessageQueues.get()) {
            update();
        }

        List<MessageQueue> messageQueues = targetMessageQueues.get();

        return messageQueues.get((int)(cursor.getAndIncrement() % messageQueues.size()));
    }
}

@@ -308,108 +308,25 @@
 ent.
-*;%0Aimport java.util.concurrent.atomic.AtomicLong;%0Aimport java.util.concurrent.atomic.AtomicReference
+ConcurrentHashMap
 ;%0A%0Ap
@@ -443,246 +443,96 @@
 ate 
-final AtomicReference%3CList%3CMessageQueue%3E%3E targetMessageQueues = new AtomicReference%3C%3E();%0A%0A    private final AtomicReference%3CList%3CMessageQueue%3E%3E messageQueues = new AtomicReference%3C%3E();%0A%0A    private final AtomicLong cursor = new AtomicLong
+ConcurrentHashMap%3CMessageQueue, Integer%3E messageQueueRegionMap = new ConcurrentHashMap%3C%3E
 ();%0A
@@ -708,32 +708,135 @@
 arget;%0A        %7D
+%0A    %7D%0A%0A    @Override%0A    public MessageQueue select(List%3CMessageQueue%3E mqs, Message msg, Object arg) %7B
 %0A%0A        if (Re
@@ -844,29 +844,24 @@
 ion.SAME == 
-this.
 target) %7B%0A  
@@ -862,37 +862,32 @@
 ) %7B%0A            
-this.
 target = Region.
@@ -961,245 +961,72 @@
     
-schedule();%0A    %7D%0A%0A    private void schedule() %7B%0A        ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() %7B%0A            @Override%0A            public Thread newThread(Runnable runnable) %7B
+if (Region.ANY == target) %7B%0A            return Util.random(mqs);
 %0A   
@@ -1030,24 +1030,27 @@
         
+%7D%0A%0A
         
 Thread t
@@ -1045,436 +1045,98 @@
     
-Thread thread = new Thread(runnable, %22SelectMessageQueueByRegion%22);%0A                thread.setDaemon(true);%0A                return thread;%0A            %7D%0A        %7D);%0A%0A        scheduler.scheduleWithFixedDelay(new Runnable() %7B%0A            @Override%0A            public void run() %7B%0A                update();%0A            %7D%0A        %7D, 30, 30, TimeUnit.SECONDS);%0A    %7D%0A%0A    private void update() %7B%0A        if (Region.ANY == this.target
+List%3CMessageQueue%3E list = new ArrayList%3C%3E();%0A%0A        for (MessageQueue messageQueue : mqs
 ) %7B%0A
@@ -1139,39 +1139,38 @@
 ) %7B%0A            
-targetM
+if (!m
 essageQueues.set
@@ -1168,55 +1168,46 @@
 ueue
-s.set(messageQueues.get());%0A            return;
+RegionMap.containsKey(messageQueue)) %7B
 %0A   
@@ -1207,35 +1207,32 @@
 ) %7B%0A        
-%7D%0A%0A
         
 List%3CMessage
@@ -1211,38 +1211,33 @@
                 
-List%3CM
+m
 essageQueue%3E all
@@ -1235,175 +1235,128 @@
 ueue
-%3E all = messageQueues.get();%0A        List%3CMessageQueue%3E result = new ArrayList%3C%3E();%0A        for (MessageQueue messageQueue : all) %7B%0A            if (Util.getRegionIndex
+RegionMap.put(messageQueue, Util.getRegionIndex(messageQueue));%0A            %7D%0A%0A            if (messageQueueRegionMap.get
 (mes
@@ -1406,21 +1406,19 @@
         
-resul
+lis
 t.add(me
@@ -1471,14 +1471,11 @@
 if (
-!resul
+lis
 t.is
@@ -1501,486 +1501,75 @@
     
-targetMessageQueues.set(result);%0A        %7D else %7B%0A            targetMessageQueues.set(all);%0A        %7D%0A    %7D%0A%0A    @Override%0A    public MessageQueue select(List%3CMessageQueue%3E mqs, Message msg, Object arg) %7B%0A        messageQueues.set(mqs);%0A%0A        if (null == targetMessageQueues.get()) %7B%0A            update();%0A        %7D%0A%0A        List%3CMessageQueue%3E messageQueues = targetMessageQueues.get();%0A%0A        return messageQueues.get((int)(cursor.getAndIncrement() %25 messageQueues.size())
+return Util.random(mqs);%0A        %7D%0A%0A        return Util.random(list
 );%0A 

Revert message queue selector.