java - 同じコマンド属性を変更する複数のスレッド

Question

問題文：-

以下のプログラムでは、を使用ThreadPoolExecutorしていArrayBlockingQueueます。

各スレッドはUNIQUE ID毎回使用する必要があり、実行する60 minutes or more必要があり60 minutesます。そのID's will get finishedため、これらのIDを再利用する必要があります。だから私はArrayBlockingQueueここでコンセプトを使用しています。

2つのシナリオ：-

command.getDataCriteria()が含まれている場合、Previous各スレッドは常にそれを使用UNIQUE ID between 1 and 1000して解放し、再利用する必要があります。
それ以外の場合、command.getDataCriteria()が含まれている場合、New各スレッドは常にそれを使用UNIQUE ID between 2000 and 3000して解放し、再利用する必要があります。

以下のプログラムで現在直面している問題-

私が直面している問題の1つは

run methodもしそうなら、それcommand.getDataCriteria()はまた、正しく起こってはならないものPreviousに入力されますか？else if block(which is for New)そしてまた私はやってい.equals checkますか？なぜこれが起こっているのですか？多くのスレッドが同時に起動し、以前にコマンドを変更したことが原因である可能性がありますか？

else if(command.getDataCriteria().equals("New")) {

マルチスレッドがそれを変更している場合、どうすればこの問題を克服できますか？どんな問題が起こっていても、それはで起こっていrun methodます。私は長い間これに固執しているので、どんな提案も大いに役立つでしょう。たぶん私たちはする必要がありますSynchronize the threads, so that no other thread should modify the command when another thread is trying to execute it.

public synchronized void runNextCommand() {

LinkedList<Integer> availableExistingIds = new LinkedList<Integer>();
LinkedList<Integer> availableNewIds = new LinkedList<Integer>();

executorService = new ThreadPoolExecutor(noOfThreads, noOfThreads, 500L, TimeUnit.MILLISECONDS,
        new ArrayBlockingQueue<Runnable>(noOfThreads), new ThreadPoolExecutor.CallerRunsPolicy());

    // If there are any free threads in the thread pool
    if (!(((ThreadPoolExecutor) executorService).getActiveCount() < noOfThreads))
    return;

    for (int i = 1; i <= 1000; i++) {
    availableExistingIds.add(i);
    }

    for (int n = 2000; n <= 3000; n++) {
    availableNewIds.add(n);
    }

    BlockingQueue<Integer> existIdPool = new ArrayBlockingQueue<Integer>(1000, false, availableExistingIds);
    BlockingQueue<Integer> newIdPool = new ArrayBlockingQueue<Integer>(1001, false, availableNewIds);

    // Running for particular duration of time
    while(System.currentTimeMillis() <= endTime) {
    Command nextCommand = getNextCommandToExecute();
    Task nextCommandExecutorRunnable = new Task(nextCommand, existIdPool, newIdPool);
    executorService.submit(nextCommandExecutorRunnable);
    }

    executorService.shutdown();
    if (!executorService.awaitTermination(Long.MAX_VALUE, TimeUnit.DAYS)) {
    executorService.shutdownNow();
    }
}

runnable（実ユニットレベルのコマンドエグゼキュータ）の実装

private static final class Task implements Runnable {
private Command command;
private DPSclient m_DPSclient = null;
private DPSclient psc = null;
private BlockingQueue<Integer> existPool;
private BlockingQueue<Integer> newPool;
private int existId;
private int newId;
private static Object syncObject = new Object();  


public Task(Command command, BlockingQueue<Integer> pool1, BlockingQueue<Integer> pool2) {
    this.command = command;
    this.existPool = pool1;
    this.newPool = pool2;
}

public void run() {

  synchronized(syncObject) {
    if(command.getDataCriteria().equals("Previous")) {
    try {
        // Getting existing id from the existPool
        existId = existPool.take();
        attributeGetSetMethod(existId);
    } catch (Exception e) {
        getLogger().log(LogLevel.ERROR, e.getLocalizedMessage());
    } finally {
        // And releasing that existing ID for re-use
        existPool.offer(existId);       
    }
} else if(command.getDataCriteria().equals("New")) {
     try {
        // Getting new id from the newPool
        newId = newPool.take();
        attributeGetSetMethod(newId);
    } catch (Exception e) {
        getLogger().log(LogLevel.ERROR, e.getLocalizedMessage());
    } finally {
        // And releasing that new ID for re-use
        newPool.offer(newId);   
    }
    }
}
  }
}

これにご協力いただきありがとうございます。ありがとう

getNextCommandToExecute method更新-マットによって提案されたコード

// Get the next command to execute based on percentages
    private synchronized Command getNextCommandToExecute() {
    int commandWithMaxNegativeOffset = 0; // To initiate, assume the first one has the max negative offset
    if (totalExecuted != 0) {
        // Manipulate that who has max negative offset from its desired execution
        double executedPercentage = ((double)executedFrequency[commandWithMaxNegativeOffset] / (double)totalExecuted) * 100;
        double offsetOfCommandWithMaxNegative = executedPercentage - commands.get(commandWithMaxNegativeOffset).getExecutionPercentage();

        for (int j=1; j < commands.size(); j++) {
        double executedPercentageOfCurrentCommand = ((double)executedFrequency[j] / (double)totalExecuted) * 100;
        double offsetOfCurrentCommand = executedPercentageOfCurrentCommand - commands.get(j).getExecutionPercentage();

        if (offsetOfCurrentCommand < offsetOfCommandWithMaxNegative) {
            offsetOfCommandWithMaxNegative = offsetOfCurrentCommand;
            commandWithMaxNegativeOffset = j;
        }
        }
    }

    // Next command to execute is the one with max negative offset
    executedFrequency[commandWithMaxNegativeOffset] ++;
    totalExecuted ++;

    // This is for User Logging/No User Logging and Data is Previous/New
    LinkedHashMap<String, Double> dataCriteriaMap = (LinkedHashMap<String, Double>) sortByValue(commands.get(commandWithMaxNegativeOffset).getDataUsageCriteria());
    Set<Map.Entry<String, Double>> entriesData = dataCriteriaMap.entrySet();
    Iterator<Map.Entry<String, Double>> itData = entriesData.iterator();
    Map.Entry<String, Double> firstEntryData = itData.next();
    Map.Entry<String, Double> secondEntryData = itData.next();

    LinkedHashMap<Boolean, Double> userCriteriaMap = (LinkedHashMap<Boolean, Double>) sortByValue(commands.get(commandWithMaxNegativeOffset).getUserLoggingCriteria());
    Set<Map.Entry<Boolean, Double>> entriesUser = userCriteriaMap.entrySet();
    Iterator<Map.Entry<Boolean, Double>> itUser = entriesUser.iterator();
    Map.Entry<Boolean, Double> firstEntryUser = itUser.next();
    Map.Entry<Boolean, Double> secondEntryUser = itUser.next();

    double percent = r.nextDouble() * 100;

    if (percent < secondEntryData.getValue().doubleValue()) {
        commands.get(commandWithMaxNegativeOffset).setDataCriteria(secondEntryData.getKey());
    } else {
        commands.get(commandWithMaxNegativeOffset).setDataCriteria(firstEntryData.getKey());
    }

    if (percent < secondEntryUser.getValue().doubleValue()) {
        commands.get(commandWithMaxNegativeOffset).setUserLogging(secondEntryUser.getKey());
    } else { 
        commands.get(commandWithMaxNegativeOffset).setUserLogging(firstEntryUser.getKey());
    }

    return commands.get(commandWithMaxNegativeOffset);
    }

そして、コマンドはクラスのトップで次のように宣言されています-

private static List<Command> commands;

もう1つの方法を更新します：-

private synchronized void attributeGetSetMethod(int id_range) {

        requestlTransaction requestlTransaction = null;
        try {
        GUID_VALUES = new LinkedHashMap<Integer, String>();

        // I am not sure how CAL logging has to be done, it has to be at each attribute level or something else? So that is the reason I left this thing.

        if(!(command.getAttributeIDSet().isEmpty())) {

            requestlTransaction = requestlTransactionFactory.create("DPSLnPTest");
            m_DPSclient = setupDPS(command.getName(), getDPSAttributeKeys(command.getDataCriteria(), command.getUserLogging() , id_range));


            for(String attr: command.getAttributeIDSet()) {

            requestlTransaction.setName("DPSAttributeSet");
            requestlTransaction.setStatus("0");
            //requestlTransaction.addData("IpAddress", ipAddress);

            if(attr.contains("/")) {
                lengthOfString = Integer.parseInt(attr.split("/")[1]);
                attr = attr.split("/")[0];
            }
            DPSAttribute attr1 = new DPSAttribute();
            attr1.setRequestAttributeId(new DPSAttributeId(Integer.parseInt(attr)));
            DPSMetadataMgr mgr = DPSMetadataMgr.getInstance();
            DPSRequestAttributeMetadata metadata = mgr.getRequestAttributeMetadataById(Integer.parseInt(attr));
            int maxOccurs = metadata.getMaxOccurs();
            String dataType = metadata.getAttributeTypeAlias();

            DPSAttributeValue attrValue1 = getRequestAttribute(dataType, lengthOfString);

            if(maxOccurs>1) {
                DPSListAttributeValue listAttrValue = new DPSListAttributeValue();
                List<DPSAttributeValue> list = new ArrayList<DPSAttributeValue>();
                list.add(attrValue1);
                listAttrValue.setList(list);
                attr1.setRequestAttributeValue(listAttrValue);
                m_DPSclient.setDPSAttribute(attr1);
            } else {
                attr1.setRequestAttributeValue(attrValue1);         
                m_DPSclient.setDPSAttribute(attr1);
            }
            }

            List<DPSAttribute> idKeys = m_DPSclient.release(PersistenceEnum.COMMIT, false);

            // Iterating through the keys and storing into HashMap
            Iterator<DPSAttribute> i = idKeys.iterator();
            while (i.hasNext()) {
            DPSAttribute DPSAttribute = (DPSAttribute)(i.next());
            DPSAttributeId id = DPSAttribute.getAttributeId();
            DPSAttributeValue value = DPSAttribute.getRequestAttribute();

            if(id.getId() == DPSLnPConstants.CGUID_ID && (value)!= null) {
                DPSLnPConstants.CGUID_VALUE = ((DPSStringAttributeValue)value).getValue();
                GUID_VALUES.put(DPSLnPConstants.CGUID_ID, DPSLnPConstants.CGUID_VALUE);
            } else if(id.getId() == DPSLnPConstants.SGUID_ID && (value)!= null) {
                DPSLnPConstants.SGUID_VALUE = ((DPSStringAttributeValue)value).getValue();
                GUID_VALUES.put(DPSLnPConstants.SGUID_ID, DPSLnPConstants.SGUID_VALUE);
            } else if(id.getId() == DPSLnPConstants.PGUID_ID && (value)!= null) {
                DPSLnPConstants.PGUID_VALUE = ((DPSStringAttributeValue)value).getValue();
                GUID_VALUES.put(DPSLnPConstants.PGUID_ID, DPSLnPConstants.PGUID_VALUE);
            } else if(id.getId() == DPSLnPConstants.UID_ID && (value)!= null) {
                DPSLnPConstants.UID_VALUE = String.valueOf(((DPSLongAttributeValue)value).getValue());
                GUID_VALUES.put(DPSLnPConstants.UID_ID, DPSLnPConstants.UID_VALUE);
            } else if(id.getId() == DPSLnPConstants.SITE_ID && (value)!= null) {
                DPSLnPConstants.SITEID_VALUE = String.valueOf(((DPSIntAttributeValue)value).getValue());
                GUID_VALUES.put(DPSLnPConstants.SITE_ID, DPSLnPConstants.SITEID_VALUE);
            } else if(id.getId() == DPSLnPConstants.ALOC_ID && (value)!= null) {
                DPSLnPConstants.ALOC_VALUE = ((DPSStringAttributeValue)value).getValue();
                GUID_VALUES.put(DPSLnPConstants.ALOC_ID, DPSLnPConstants.ALOC_VALUE);
            } else if(id.getId() == DPSLnPConstants.ULOC_ID && (value)!= null) {
                DPSLnPConstants.ULOC_VALUE = ((DPSStringAttributeValue)value).getValue();
                GUID_VALUES.put(DPSLnPConstants.ULOC_ID, DPSLnPConstants.ULOC_VALUE);
            } else if(id.getId() == DPSLnPConstants.SLOC_ID && (value)!= null) {
                DPSLnPConstants.SLOC_VALUE = ((DPSStringAttributeValue)value).getValue();
                GUID_VALUES.put(DPSLnPConstants.SLOC_ID, DPSLnPConstants.SLOC_VALUE);
            } else if(id.getId() == DPSLnPConstants.PLOC_ID && (value)!= null) {
                DPSLnPConstants.PLOC_VALUE = ((DPSStringAttributeValue)value).getValue();
                GUID_VALUES.put(DPSLnPConstants.PLOC_ID, DPSLnPConstants.PLOC_VALUE);
            }
            }

            // Storing all the locators, guid in a map corresponding to an ID, then later on insert everything directly into db
            GUID_ID_MAPPING.put(id_range, GUID_VALUES);

            // Sleeping the command for particular milliseconds
            // One thing not sure, I should be sleeping the command here or I should put it above this comment line '// Iterating through the keys'
            Thread.sleep(command.getSleepTime());
        } 

        // for get attributes   
        // And also how CAL logging has to be done here too. And we can use same DPS Smart Client that got created above to get the attributes value?

        if(!(command.getAttributeIDGet().isEmpty())) {

            requestlTransaction.setName("DPSAttributeGet");
            requestlTransaction.setStatus("1");
            psc = setupDPS(command.getName(), getDPSAttributeKeys(command.getDataCriteria(), command.getUserLogging() , id_range));

            for(String attr: command.getAttributeIDGet()) {
            DPSAttribute attribute = new DPSAttribute();
            attribute = psc.getDPSAttribute(new DPSAttributeId(Integer.parseInt(attr)));
            //System.out.println(attribute.getAttributeId()+ " " +attribute.getRequestAttribute());
            }
        }
        } catch(Exception e) {
        getLogger().log(LogLevel.ERROR, e);
        } finally {
        requestlTransaction.completed();
        }

    }

score 1 · Accepted Answer

1つのスレッドだけがアクセスすることになっている場合command.getDataCriteria()は、スレッドの同期の問題が原因でその値が変更されないことがわかります。これを想定して、2つのスレッドがオブジェクトインスタンスへのハンドルを取得しないようにします。

あなたは持っていますが、あなたの定義に基づいてsynchronized getNextCommandToExecute()あなたのコレクションを同期させる証拠はありません。上記の仮定が成り立つ場合、2つのスレッドがコレクションから同じオブジェクトインスタンスを作成できないことを保証します。commandsprivate static List<Command> commands;get()

コードの他のどこcommandsからコレクションにアクセスしていますか？

適切に同期できる場合commands.get()は、返されたインスタンスでスレッドの競合が発生しないようにする必要があります。どちらのスレッドがオブジェクトを取得しても、それを所有します。

...私が間違った木を吠えていると私に言ってください

コメントをフォローして[編集]

私はすべてのコードを持っているわけではなく、仮定があるので、何が悪いのかを確実に言うのは難しいです。synchronized多くのメソッドにキーワードを追加していて、問題が解決することを期待しているように見えます。最善のアプローチは、実際に同期が必要なコード行を最小限に抑えることであり、これには、実際に同期する必要があるものをより明確に理解する必要があります。

スレッド間でオブジェクトへの参照を共有したくないCommandので、get（）操作とadd（）操作List<Command> commandsが同期していることを確認してください。あなたはsynchronizedListまたはおそらくConcurrentLinkedQueueの使用を採用することができます
Commandオブジェクトを不変にすることはできますか？（つまり、setX（）メソッドはありません）

java - 同じコマンド属性を変更する複数のスレッド

1 に答える 1

Related

Reference