java - 消费者和 worker 模式

我有10名工人清单的代码，以及以下两种方法：

public void demoDeques() {
        int maxSizeOfJobDeque = 3;
        Producer producer = new ProducerImpl( maxSizeOfJobDeque );

        Logger.debug( "WorkFlowEngineImpl : " +
                "Creating Workers and adding them to allocator" );
        List<Worker> workerList = buildWorkerList( producer );
        Logger.debug( "WorkFlowEngineImpl : " +
                "Assigning some jobs to the workers. " +
                    "The workers have not been started yet");

        for ( int i=1; i<4; i++ ) {
            producer.assign( new JobImpl( "job " + i, i ) );
            try {
                Thread.sleep( 4000 );
            } catch( InterruptedException e ) {
                e.printStackTrace();
            }
        }

        Logger.debug( "WorkFlowEngineImpl : " + "Starting the workers" );
        startWorkersAndWait5Seconds( workerList );
        Logger.debug( "WorkFlowEngineImpl : " +
                "Assigning some more jobs to the " +
                    "started workers" );

        for ( int i=4; i<7; i++ ) {
            producer.assign( new JobImpl( "Job " + i, i ) );
            try {
                Thread.sleep( 4000 );
            } catch( InterruptedException e ) {
                e.printStackTrace();
            }
        }

        Logger.debug( "WorkFlowEngineImpl : " + "Assigning More Jobs" );
        for ( int i=7; i<11; i++ ) {
            producer.assign( new JobImpl( "job" + i, i ) );
            try {
                Thread.sleep( 4000 );
            } catch( InterruptedException e ) {
                e.printStackTrace();
            }
        }
    }

生产者：

public synchronized void assign( Job job ) {
    Set<Worker> workerSet = jobMap.keySet();
    LinkedBlockingDeque<Job> jobQueue;
    StringBuffer sb;

    for ( Worker worker : workerSet ) {
        jobQueue = jobMap.get( worker );

        sb = new StringBuffer();
        sb.append( "Assigning job " );
        sb.append( job.getJobNumber() );
        sb.append( " to " );
        sb.append( worker );
        sb.append( "'s jobs Deque" );

        Logger.debug( "Producer : " + sb.toString() );

        if ( ! jobQueue.offerFirst( job ) ) {
            jobQueue.pollLast();
            jobQueue.offerFirst( job );
        }
    }
}

我正在尝试更改这两种方法，以便让分配器具有100个作业的列表，然后进行分配，以使十个工人中的每个工人最多可以同时获得三个作业，直到达到100个，即工人一个作业1,2,3工人2占用4,5,6，这样当工人10到达时，它又回到工人1分配三个作业，直到达到第100个作业，然后它停止并提醒所有作业已分配。请帮助我被卡住...

最佳答案

尽管您的问题提到了特定的执行顺序，但您的评论似乎表明这不是真正的要求。建议不要使用Java 1.5中提供的Executors来实现自己的线程池。例如：

ExecutorService threadPool =
    Executors.newFixedThreadPool(NUMBER_THREADS_TO_RUN_JOBS);
// your producers then just have to submit jobs to the pool
for ( int i=1; i<4; i++ ) {
    threadPool.submit(new JobImpl("Job " + i, i));
}

我无法从您的示例中确切看到每个“作业”需要进行哪些处理，但是如果JobImpl需要实现Runnable才能使上述代码正常工作。如果需要从工作中返回一些值，则可以将JobImpl更改为Callable，并使用Future方法返回的submit(Callable)来获取call()返回的结果。