java - Spring 同步问题

Question

我知道关于重用原型(prototype) bean 的问题被问过很多次，但我的问题不止于此。

问题是什么:

我在 for-lookup 的处理程序(原型(prototype) bean)中启动异步任务。但我无法在前一个达到某个里程碑之前开始下一个异步任务。所以只有在前一个任务调用了一个特殊方法后，我才必须进行for循环。

什么问题:

1. 如何在另一个 bean 中调用某些方法之前在 for 循环中等待？
2. 我可以从另一个 bean 调用当前原型(prototype) bean 的 proceedLookUp() 方法吗？

---

@Service(value = "Request.Start")
@Scope("prototype")
public static class Start {

  public Start() {}

  private Object lock;

  @Transactional
  public void handler(Request request, Response response) {

    for (int i = 0; i < request.getAmount(); i++) {
      Utils.asyncProcessStart(); //Can't start the next async process before the previous rich some defined milestone
      lock.wait();
    }
  }

  public void proceedLookUp() {
    lock.notify();
  }
}

@Service
public void AsynchronousTask {

  public void asyncAction() [
    //Needed logic, before start the next async task
    getStartHandler().proceedLookUp();
  }

  public void getStartHandler() {
    //HOW TO REWRITE NEEDED PROTOTYPE BEAN
  } 
}

添加:

问题是什么:我用的是Activiti框架，有一些限制。我应该将一些变量存储到进程(线程)上下文中。我可以将变量写入全局上下文，但不能在进程(线程)启动之前写入本地进程(线程)上下文。

what you expect to happen, say if request.getAmount() returns 2?

我应该在两个不同的线程中启动两个异步进程。每个进程都有相同的变量集。我必须将适当的变量写入每个进程(线程)的本地上下文。但是，我不能在进程(线程)启动之前(由于 Activiti 框架的特殊性)。

例如，每个进程(线程)都应该将“id”属性写入自己的本地上下文。我在 handler 方法中有 List ids

所以，我应该执行下一个操作序列:

1. 将 ids.get(0) 作为“id”属性写入全局上下文
2. 开始第一个过程
3. 坚持for循环
4. [在第一个进程内] 将“id”属性从全局写入本地上下文(在第一个进程内是可能的)
5. 通知适当的 Start bean 它可以继续 for 循环
6. 将 ids.get(1) 作为“id”属性写入全局上下文
7. 开始第二个过程
8. 坚持for循环
9. [在第二个进程中] 将“id”属性从全局写入本地上下文(在第二个进程中是可能的)
10. 通知适当的 Start bean 它可以继续 for 循环

Why can't you just call it synchronously?

正如您已经了解的那样，无法保证第一个进程(线程)在被第二个进程(线程)的 for 循环覆盖之前将“id”属性写入其本地上下文。

Answer 1

这是我的建议:创建一个单例对象，您可以在线程之间共享该对象以传递信息(状态)。单例使用 semaphore协调你的线程。您可以使用这种方法将新线程的身份传递回您的服务类。这是一个简单的例子，展示了我的建议。

测试类:

公共(public)类测试信号量{

    @Test
    public void test() throws Exception {

        ThreadCoordinator tc = ThreadCoordinator.getInstance();

        for( int i = 0; i < 100; i++ ) {
            MyThread r = new MyThread();
            r.run();

            // This will block until the Thread has called release (after setting its identity on the ThreadCoordinator)    
            tc.acquire();
            String newThreadIdentity = tc.getIdentity();
            System.out.println( "Received the new thread's identity:         " + newThreadIdentity );

            // This will allow the next Thread to acquire the semaphore
            tc.release();
        }
    }


    class MyThread extends Thread {

        public void run() {
            String identity = Integer.toString( (int)(Math.random() * 10000) );
            System.out.println( "Running a new thread with identity:         " + identity );

            // Get a reference to the singleton
            ThreadCoordinator tc = ThreadCoordinator.getInstance();
            try {
                tc.acquire();
                tc.setIdentity( identity );
                System.out.println( "Notified the ThreadCoordinator from thread: " + identity );
                tc.release();
            } catch( InterruptedException e ) {
                System.out.println( "Caught an interrupted exception: " + e );
            }

        }

    }
}

ThreadCoordinator(信号量)类:

import java.util.concurrent.Semaphore;

public class ThreadCoordinator {

        private static ThreadCoordinator tc = new ThreadCoordinator();
        private static Semaphore semaphore = new Semaphore( 1, true );      
        private static String identity;

        // singleton get instance
        public static ThreadCoordinator getInstance() {
            return ThreadCoordinator.tc;
        }

        public void setIdentity( String identity ) throws InterruptedException {
            ThreadCoordinator.identity = identity;
        }


        public String getIdentity() throws InterruptedException {
            String identity = ThreadCoordinator.identity;
            ThreadCoordinator.identity = null;
            return identity;
        }

        public void acquire() throws InterruptedException {
            ThreadCoordinator.semaphore.acquire();
        }

        public void release() {
            ThreadCoordinator.semaphore.release();
        }

 }