Java8+多Map并行查找ip-address并终止任务的方案咨询

先梳理你之前方案的问题 #

• RxJava方案：未指定调度器，默认使用当前线程串行执行，无法实现并行查找
• 自定义线程方案：每个线程内的AtomicReference是局部变量，并非多线程共享；同时没有中断/终止逻辑，导致所有线程都会完成查找操作

以下是三种可行的实现方案：

方案一：Java8原生CompletableFuture（推荐） #

利用CompletableFuture的异步能力和anyOf方法获取第一个完成的任务结果，同时取消未完成的任务，避免无效计算。

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

public class MapIpLookup {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 初始化测试Map
        Map<String, String> map1 = new HashMap<>();
        map1.put("net-ip", "333");
        map1.put("ip-address", "132333");

        Map<String, String> map2 = new HashMap<>();
        map2.put("hostname", "kekdkd");
        map2.put("ip-address", "232233");
        map2.put("operating-system", "win11");

        Map<String, String> map3 = new HashMap<>();
        map3.put("processor", "intel");
        map3.put("ip-address", "344322");
        map3.put("graphics", "nvidia");

        // 创建并行查找任务
        CompletableFuture<String> task1 = CompletableFuture.supplyAsync(() -> lookupIp(map1));
        CompletableFuture<String> task2 = CompletableFuture.supplyAsync(() -> lookupIp(map2));
        CompletableFuture<String> task3 = CompletableFuture.supplyAsync(() -> lookupIp(map3));

        // 获取第一个完成的任务结果
        CompletableFuture<Object> firstCompleted = CompletableFuture.anyOf(task1, task2, task3);
        String result = (String) firstCompleted.get();
        System.out.println("找到的IP地址：" + result);

        // 取消未完成的任务
        task1.cancel(true);
        task2.cancel(true);
        task3.cancel(true);
    }

    private static String lookupIp(Map<String, String> map) {
        // 检查任务是否被取消，及时终止
        if (Thread.currentThread().isInterrupted()) {
            return null;
        }
        // 直接用Map.get()比遍历entrySet高效
        String ip = map.get("ip-address");
        // 模拟耗时操作（测试用）
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return null;
        }
        return ip;
    }
}

方案二：修正RxJava3并行实现 #

通过指定调度器让任务并行执行，利用firstElement()自动终止其他流的订阅。

import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.schedulers.Schedulers;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.atomic.AtomicReference;

public class RxJavaIpLookup {
    public static void main(String[] args) {
        Map<String, String> map1 = new HashMap<>();
        map1.put("net-ip", "333");
        map1.put("ip-address", "132333");

        Map<String, String> map2 = new HashMap<>();
        map2.put("hostname", "kekdkd");
        map2.put("ip-address", "232233");

        Map<String, String> map3 = new HashMap<>();
        map3.put("processor", "intel");
        map3.put("ip-address", "344322");

        AtomicReference<String> ipAddress = new AtomicReference<>();

        Observable.merge(
                getIpFromMap(map1).subscribeOn(Schedulers.io()),
                getIpFromMap(map2).subscribeOn(Schedulers.io()),
                getIpFromMap(map3).subscribeOn(Schedulers.io())
        )
        .firstElement() // 取第一个结果，自动取消其他Observable
        .subscribe(ip -> {
            ipAddress.set(ip);
            System.out.println("找到的IP地址：" + ip);
        });

        // 测试用：等待结果返回
        try {
            Thread.sleep(500);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    private static Observable<String> getIpFromMap(Map<String, String> map) {
        return Observable.defer(() -> {
            if (Thread.currentThread().isInterrupted()) {
                return Observable.empty();
            }
            String ip = map.get("ip-address");
            // 模拟耗时操作
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                return Observable.empty();
            }
            return Observable.just(ip);
        });
    }
}

方案三：自定义线程+共享中断标记 #

修正你之前的多线程方案，使用共享标记和线程中断机制实现终止逻辑。

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;

public class ThreadIpLookup {
    // 共享的"已找到"标记
    private static final AtomicBoolean found = new AtomicBoolean(false);
    private static String resultIp;

    public static void main(String[] args) throws InterruptedException {
        Map<String, String> map1 = new HashMap<>();
        map1.put("net-ip", "333");
        map1.put("ip-address", "132333");

        Map<String, String> map2 = new HashMap<>();
        map2.put("hostname", "kekdkd");
        map2.put("ip-address", "232233");

        Map<String, String> map3 = new HashMap<>();
        map3.put("processor", "intel");
        map3.put("ip-address", "344322");

        Thread t1 = new Thread(() -> lookup(map1), "t1");
        Thread t2 = new Thread(() -> lookup(map2), "t2");
        Thread t3 = new Thread(() -> lookup(map3), "t3");

        t1.start();
        t2.start();
        t3.start();

        // 等待第一个线程找到结果
        synchronized (ThreadIpLookup.class) {
            ThreadIpLookup.class.wait();
        }

        // 中断剩余线程
        t1.interrupt();
        t2.interrupt();
        t3.interrupt();

        System.out.println("找到的IP地址：" + resultIp);
    }

    private static void lookup(Map<String, String> map) {
        // 先检查是否已找到，避免无效执行
        if (found.get()) {
            return;
        }

        try {
            // 模拟耗时操作
            Thread.sleep(100);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return;
        }

        // 检查线程是否被中断
        if (Thread.currentThread().isInterrupted()) {
            return;
        }

        String ip = map.get("ip-address");
        // CAS操作确保只有第一个线程能设置结果
        if (found.compareAndSet(false, true)) {
            resultIp = ip;
            // 通知主线程
            synchronized (ThreadIpLookup.class) {
                ThreadIpLookup.class.notify();
            }
        }
    }
}

内容的提问来源于stack exchange，提问作者astar