我审查了多线程,并尝试实现一个创建单独线程来运行收集进程的应用程序。该过程中的主要方法需要一个变量数组列表,我正在尝试找出一种将数组列表传递给每个线程的方法。
ApplicationContext context = new AnnotationConfigApplicationContext(AppConfig.class);
ThreadPoolTaskExecutor taskExecutor = (ThreadPoolTaskExecutor) context.getBean("taskExecutor");
MainTask mxTask = (MainTask) context.getBean("MainTask");
mxTask.setName("Thread 1");
taskExecutor.execute(mxTask);
MainTask mxTask2 = (MainTask) context.getBean("MainTask");
mxTask2.setName("Thread 2");
taskExecutor.execute(mxTask2);
上面是声明线程的类方法,MainTask 是执行 run() 方法的类,然后调用其他 main 方法。
@Override
public void run() {
System.out.println(name + " is running.");
getConfigurations();
try {
mainRun();
} catch (MessagingException e1) {
// TODO Auto-generated catch block
e1.printStackTrace();
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
getconfigurations() 方法读取所有必要的配置变量。下面是每个线程需要运行的主要流程。
public static void mainRun() throws IOException {
ArrayList<String> filterList = new ArrayList<String>();
ArrayList<String> brokerList = new ArrayList<String>();
if(kafkaServers1 != null) {
brokerList.add(kafkaServers1);
} else if(!kafkaServers2.isEmpty()) {
brokerList.add(kafkaServers2);
} else if(!kafkaServers3.isEmpty()) {
brokerList.add(kafkaServers3);
}
ArrayList<String> serverList = new ArrayList<String>();
for(int x=0;x<brokerList.size();x++){
String[] serverBrokers = brokerList.get(x).split(",");
serverList.add(serverBrokers[0]);
serverList.add(serverBrokers[1]);
serverList.add(serverBrokers[2]);
}
try {
while(true){
for (String temp : serverList) {
kafkaServer = temp;
hostName = kafkaServer;
InetAddress addr = InetAddress.getByName(hostName);
hostName = addr.getHostName();
kafkaServer= hostName;
retrieveData(hostName);
...
变量 kafkaServers1 包含 3 个 ip 的列表,这些 ip 被拆分并分别添加到 serverList arrayList 中。我想做的只是为每个线程分配一个IP。这可能吗?有人可以建议一下吗?
最佳答案
您可以检索 Kafka 配置,然后为每个主机创建一个新的 Runnable,而不是让 Runnable 对象成为一个 bean。提交给执行人。
MyRunnable infoGetter = new MyRunnable(hostInfo);
taskExecutor.execute(infoGetter);
如果您需要对结果执行某些操作,CompletableFuture 可能是更好的选择,或者 fork/join RecursiveTask<V>
如果您需要合并它们。
关于java - 将参数传递给 threadpoolexecutor,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/54828919/