Java: Threadpoolexecutor - 使用作业列表来提交作业？
Java: Threadpoolexecutor - use list of jobs to submit jobs?
下面的代码是Customer和jobManager。客户有姓名、地址和账户余额。工作是从一个客户到另一个客户的汇款。这是一个 Threadpoolexecutor 训练程序。以下版本有效，我一个一个提交作业
customer.java
public class customer {
private String name;
private String adress;
private Double accountBalance;
public customer(String name, String adress, Double accountBalance)
{
this.name = name;
this.adress = adress;
this.accountBalance = accountBalance;
}
public String getName() { return name; }
public String getAdress()
{
return adress;
}
public Double getAccountBalance(){return accountBalance;}
public void setAccountBalance(double accountBalance){this.accountBalance=accountBalance;}
@Override
public String toString(){
return "[" + name+"; " +adress+"; "+accountBalance+"]";
}
}
customerOrganizer.java
import java.util.ArrayList;
import java.util.List;
public class customerOrganizer {
private static final customerOrganizer myJobOrganizer = new customerOrganizer();
public static customerOrganizer getJobOrganizer(){
return myJobOrganizer;
}
private List<customer> customerList = new ArrayList<customer>();
public void add_customer(customer kunde)
{
this.customerList.add(kunde);
}
public Iterable<customer> all_customers()
{
return this.customerList;
}
public static customerOrganizer getInstance()
{
return myJobOrganizer;
}
}
job.java
public class job implements Runnable {
private customer kunde1;
private customer kunde2;
private Double transfer;
public job(customer kunde1, customer kunde2, Double transfer) {
this.kunde1 = kunde1;
this.kunde2 = kunde2;
this.transfer = transfer;
}
@Override
public String toString(){
return "[" + kunde1+"; " +kunde2+"; "+transfer+"]";
}
public void run() {
System.out.println("Starting transfer");
Double geber = this.kunde1.getAccountBalance();
Double nehmer = this.kunde2.getAccountBalance();
Double geberNeu = geber - this.transfer;
this.kunde1.setAccountBalance(geberNeu);
Double nehmerNeu = nehmer + this.transfer;
this.kunde2.setAccountBalance(nehmerNeu);
System.out.println("Transfer done");
}
}
jobOrganizer.java
public class jobOrganizer {
private static final jobOrganizer myJobOrganizer = new jobOrganizer();
public static jobOrganizer getMyJobOrganizer() {
return myJobOrganizer;
}
private List<job> jobList = new ArrayList<job>();
public int getAmount(){ return jobList.size();}
public void add_job(job newJob) {
this.jobList.add(newJob);
}
public Iterable<job> all_jobs() {
return this.jobList;
}
public static jobOrganizer getInstance() {
return myJobOrganizer;
}
}
Main.java
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class Main {
public static void main(String[] args) {
customerOrganizer myCustomerOrganizer = new customerOrganizer();
jobOrganizer myJobOrganizer= new jobOrganizer();
customer mueller = new customer("Tim Mueller", "Strasse 1", 1077.00);
customer john = new customer("John Doe", "Strasse 2",503.00);
customer meier = new customer("John Meier", "Strasse 3", 8500.50);
customer wurst = new customer("Hans Wurst", "Strasse 4", 1000.00);
myCustomerOrganizer.add_customer(mueller);
myCustomerOrganizer.add_customer(john);
myCustomerOrganizer.add_customer(meier);
myCustomerOrganizer.add_customer(wurst);
job transfer1= new job(meier,wurst,500.50);
job transfer2= new job(mueller,john,77.00);
myJobOrganizer.add_job(transfer1);
myJobOrganizer.add_job(transfer2);
// this works:
ExecutorService executor = Executors.newFixedThreadPool(2);
executor.submit(transfer1);
executor.submit(transfer2);
executor.shutdown();
}}
所以，我确实有一个 jobList，而且我有想法，我应该使用它。我不想一个接一个地提交作业，而是根据作业列表提交它们。一开始我想到了这样的事情：
int threads = myJobOrganizer.getAmount();
ExecutorService executor = Executors.newFixedThreadPool(threads);
for (int i = 0; i <threads+1; i++){
//submit jobs? execute?
}
此外，myJobOrganizer 需要实现 Runnable 吗？我还看到解决方案是这样的：
for(condition){
executor.execute(new Runnable() {
@Override
public void run() {
submit the jobs?
}
}}
但是我真的不知道，该怎么做。基本上，我缺少如何以正确的方式从我的 jobList 中提取作业，以便我可以将它们提交给执行程序服务 >.<
关于线程安全的更新
所以，我按照link提供的Rab，我用了一个CompletionService。 Main.java 的最后一部分现在看起来像这样：
int threads = myJobOrganizer.getAmount();
System.out.println(myCustomerOrganizer.all_customers().toString());
// executor service
ExecutorService executor = Executors.newFixedThreadPool(threads);
// completion service is applied on executor
CompletionService service = new ExecutorCompletionService(executor);
for(Callable<Job> myJob : myJobOrganizer.all_jobs()){
service.submit(myJob);
}
executor.shutdown();
// pause the main for control printout -> not nice yet, I am working on
// joining threads
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
// control print
System.out.println(myCustomerOrganizer.all_customers().toString());
请注意，此编辑旨在完成条目，但仍然是错误的（很遗憾）。提供的答案是 link原始问题的答案，不涉及线程安全
感谢您的时间和努力！
ExecutorService 处理如何在工作人员之间分配任务。你只需要一个接一个地传递任务即可。
for (job jobObj : myJobOrganizer.all_jobs())
executor.submit(jobObj);
注意sumbit return一个Future<?>，它用来跟踪你的任务是否完成，或者是否出错（还有任务结果，但是runnable没有结果）。如果您关心这些东西，您会希望将它们收集在某种容器中，例如 List.
如果把job改成Callable<Void>，提交会容易很多。 Callable 是 Runnable 的某种扩展，允许任务在完成时产生结果。由于您的传输没有结果，因此使用 java.lang.Void 作为泛型参数的填充类型就可以了。
现在，只要executor.invokeAll(myJobOrganizer.all_jobs())就足够了。这将节省一些上下文切换，加快速度。 （其实很重要，因为你的任务都很小）
顺便说一句，您应该知道并发访问需要适当的同步，您有 none。如果不同的工作涉及同一个帐户，您的帐户最终可能会处于错误状态。我们也常用LargeCamelCase命名类，smallCamelCase命名方法。
如果您不想使用循环，可以使用 Stream 来完成此操作。如果您使用的是大于或等于 8 的 java 版本，这里是一个示例
myJobList
.stream()
.forEach(e -> executor.execute(() -> {
//submit the jobs
}));
或
myJobOrganizer.all_jobs()
.stream()
.forEach(e -> executor.submit(e));
如果您真的不想循环，您可以使用 executor.invokeAll(myJobList) 提交列表。
我怎么发现这个答案很有趣你应该完全研究一下
下面的代码是Customer和jobManager。客户有姓名、地址和账户余额。工作是从一个客户到另一个客户的汇款。这是一个 Threadpoolexecutor 训练程序。以下版本有效，我一个一个提交作业
customer.java
public class customer {
private String name;
private String adress;
private Double accountBalance;
public customer(String name, String adress, Double accountBalance)
{
this.name = name;
this.adress = adress;
this.accountBalance = accountBalance;
}
public String getName() { return name; }
public String getAdress()
{
return adress;
}
public Double getAccountBalance(){return accountBalance;}
public void setAccountBalance(double accountBalance){this.accountBalance=accountBalance;}
@Override
public String toString(){
return "[" + name+"; " +adress+"; "+accountBalance+"]";
}
}
customerOrganizer.java
import java.util.ArrayList;
import java.util.List;
public class customerOrganizer {
private static final customerOrganizer myJobOrganizer = new customerOrganizer();
public static customerOrganizer getJobOrganizer(){
return myJobOrganizer;
}
private List<customer> customerList = new ArrayList<customer>();
public void add_customer(customer kunde)
{
this.customerList.add(kunde);
}
public Iterable<customer> all_customers()
{
return this.customerList;
}
public static customerOrganizer getInstance()
{
return myJobOrganizer;
}
}
job.java
public class job implements Runnable {
private customer kunde1;
private customer kunde2;
private Double transfer;
public job(customer kunde1, customer kunde2, Double transfer) {
this.kunde1 = kunde1;
this.kunde2 = kunde2;
this.transfer = transfer;
}
@Override
public String toString(){
return "[" + kunde1+"; " +kunde2+"; "+transfer+"]";
}
public void run() {
System.out.println("Starting transfer");
Double geber = this.kunde1.getAccountBalance();
Double nehmer = this.kunde2.getAccountBalance();
Double geberNeu = geber - this.transfer;
this.kunde1.setAccountBalance(geberNeu);
Double nehmerNeu = nehmer + this.transfer;
this.kunde2.setAccountBalance(nehmerNeu);
System.out.println("Transfer done");
}
}
jobOrganizer.java
public class jobOrganizer {
private static final jobOrganizer myJobOrganizer = new jobOrganizer();
public static jobOrganizer getMyJobOrganizer() {
return myJobOrganizer;
}
private List<job> jobList = new ArrayList<job>();
public int getAmount(){ return jobList.size();}
public void add_job(job newJob) {
this.jobList.add(newJob);
}
public Iterable<job> all_jobs() {
return this.jobList;
}
public static jobOrganizer getInstance() {
return myJobOrganizer;
}
}
Main.java
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class Main {
public static void main(String[] args) {
customerOrganizer myCustomerOrganizer = new customerOrganizer();
jobOrganizer myJobOrganizer= new jobOrganizer();
customer mueller = new customer("Tim Mueller", "Strasse 1", 1077.00);
customer john = new customer("John Doe", "Strasse 2",503.00);
customer meier = new customer("John Meier", "Strasse 3", 8500.50);
customer wurst = new customer("Hans Wurst", "Strasse 4", 1000.00);
myCustomerOrganizer.add_customer(mueller);
myCustomerOrganizer.add_customer(john);
myCustomerOrganizer.add_customer(meier);
myCustomerOrganizer.add_customer(wurst);
job transfer1= new job(meier,wurst,500.50);
job transfer2= new job(mueller,john,77.00);
myJobOrganizer.add_job(transfer1);
myJobOrganizer.add_job(transfer2);
// this works:
ExecutorService executor = Executors.newFixedThreadPool(2);
executor.submit(transfer1);
executor.submit(transfer2);
executor.shutdown();
}}
所以，我确实有一个 jobList，而且我有想法，我应该使用它。我不想一个接一个地提交作业，而是根据作业列表提交它们。一开始我想到了这样的事情：
int threads = myJobOrganizer.getAmount();
ExecutorService executor = Executors.newFixedThreadPool(threads);
for (int i = 0; i <threads+1; i++){
//submit jobs? execute?
}
此外，myJobOrganizer 需要实现 Runnable 吗？我还看到解决方案是这样的：
for(condition){
executor.execute(new Runnable() {
@Override
public void run() {
submit the jobs?
}
}}
但是我真的不知道，该怎么做。基本上，我缺少如何以正确的方式从我的 jobList 中提取作业，以便我可以将它们提交给执行程序服务 >.<
关于线程安全的更新
所以，我按照link提供的Rab，我用了一个CompletionService。 Main.java 的最后一部分现在看起来像这样：
int threads = myJobOrganizer.getAmount();
System.out.println(myCustomerOrganizer.all_customers().toString());
// executor service
ExecutorService executor = Executors.newFixedThreadPool(threads);
// completion service is applied on executor
CompletionService service = new ExecutorCompletionService(executor);
for(Callable<Job> myJob : myJobOrganizer.all_jobs()){
service.submit(myJob);
}
executor.shutdown();
// pause the main for control printout -> not nice yet, I am working on
// joining threads
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
// control print
System.out.println(myCustomerOrganizer.all_customers().toString());
请注意，此编辑旨在完成条目，但仍然是错误的（很遗憾）。提供的答案是 link原始问题的答案，不涉及线程安全
感谢您的时间和努力！
ExecutorService 处理如何在工作人员之间分配任务。你只需要一个接一个地传递任务即可。
for (job jobObj : myJobOrganizer.all_jobs())
executor.submit(jobObj);
注意sumbit return一个Future<?>，它用来跟踪你的任务是否完成，或者是否出错（还有任务结果，但是runnable没有结果）。如果您关心这些东西，您会希望将它们收集在某种容器中，例如 List.
如果把job改成Callable<Void>，提交会容易很多。 Callable 是 Runnable 的某种扩展，允许任务在完成时产生结果。由于您的传输没有结果，因此使用 java.lang.Void 作为泛型参数的填充类型就可以了。
现在，只要executor.invokeAll(myJobOrganizer.all_jobs())就足够了。这将节省一些上下文切换，加快速度。 （其实很重要，因为你的任务都很小）
顺便说一句，您应该知道并发访问需要适当的同步，您有 none。如果不同的工作涉及同一个帐户，您的帐户最终可能会处于错误状态。我们也常用LargeCamelCase命名类，smallCamelCase命名方法。
如果您不想使用循环，可以使用 Stream 来完成此操作。如果您使用的是大于或等于 8 的 java 版本，这里是一个示例
myJobList
.stream()
.forEach(e -> executor.execute(() -> {
//submit the jobs
}));
或
myJobOrganizer.all_jobs()
.stream()
.forEach(e -> executor.submit(e));
如果您真的不想循环，您可以使用 executor.invokeAll(myJobList) 提交列表。
我怎么发现这个答案很有趣你应该完全研究一下