java - 优化Java中的多线程队列处理代码

Question

我有一段代码创建了一个类的 10 个对象，这些对象实现了 runnable。 每个对象都保存在 hashmap 中以供以后使用。 每个对象都在单独的线程上运行。每个对象都有一个公共(public)方法，可以将项目添加到队列中。 该对象以无限循环处理队列。

我想知道这个解决方案是否可行，或者是否存在完全错误/无用/缺失的内容(尤其是使用 volatile 和 synchronized 关键字)？

MultithreadingTest.class

package multithreadingtest;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Multithreading example.
 *
 * @author lkallas
 */
public class MultithreadingTest {

    private static final int NUM_OF_THREADS = 10;
    private static String name;
    private static final Map<Integer, ThreadWorker> objectMap = new HashMap<>();    //Map or storing Threadworker objects

    public static void main(String[] args) {
        ExecutorService executor = Executors.newFixedThreadPool(NUM_OF_THREADS);
        //Creating threads
        for (int i = 0; i < NUM_OF_THREADS; i++) {
            name = "ThreadWorker" + String.valueOf(i);
            ThreadWorker thread = new ThreadWorker(name);
            objectMap.put(i, thread);   //Add objects to map            
            executor.execute(thread);
        }
        for (int i = 0; i < 10; i++) {
            ThreadWorker worker = objectMap.get(i);
            for (int j = 0; j < 10; j++) {
                worker.addToQueue("Test1");
            }
        }
    }
}

ThreadWorker.class

package multithreadingtest;

import java.util.LinkedList;
import java.util.Queue;

/**
 * Worker class that performs operations in another thread.
 *
 * @author lkallas
 */
public class ThreadWorker implements Runnable {

    private final String threadName;
    private volatile Queue workQueue;   //Does this have to volatile??

    /**
     * Class constructor.
     *
     * @param threadName Name of the thread for identifying.
     *
     */
    public ThreadWorker(String threadName) {
        this.threadName = threadName;
        this.workQueue = new LinkedList();
        System.out.println(String.format("Thread %s started!", threadName));
    }

    /**
     * Adds items to the queue.
     *
     * @param object Object to be added to the queue.
     */
    public synchronized void addToQueue(String object) {
        workQueue.add(object); //Does it have to be syncronized void
    }

    @Override
    public void run() {
        while (true) {
            if (!workQueue.isEmpty()) {
                System.out.println("Queue size: " + workQueue.size());
                String item = (String) workQueue.peek();
                //Process item
                System.out.println(threadName + " just processed " + item);
                workQueue.remove();
            }
        }
    }
}

非常感谢任何帮助和建议!

Answer 1

1. workQueue是线程本地的，不需要 volatile (它是私有(private)的，没有公共(public) setter 方法)
2. 制造workQueue一个BlockingQueue - 这个队列是线程安全的，所以你不需要同步addToQueue 。另外，你不需要在run内部旋转。 - 相反，您可以调用take()在队列中，线程会阻塞，直到有一个项目可用为止。
3. 您似乎在 MultithreadingTest 内做了太多工作- 您可以让所有工作人员共享相同的BlockingQueue，而不是将项目添加到单独的队列中，然后main只需要向该单条 BlockingQueue 添加项目即可并且工作人员将自行负责负载平衡。请注意，即使 BlockingQueue是共享的，还是不需要 volatile因为对 BlockingQueue 的引用一旦工作进程初始化就不会改变(使字段 private final BlockingQueue<String> workQueue - final 字段永远不需要是 volatile )。