代码之家 › 专栏 › 技术社区 › lapots

使用threadfactory自定义线程命名

concurrency java

-1

lapots · 技术社区 · 6 年前

world

public class WorldLifecycle {
    private ExecutorService executorService;

    public void startWorld() {
        executorService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors(),
            r -> {
                String id = ((IdentifiableRunnable) r).getId();
                return new Thread(r, id);
            });
    }

    public void bringLife(LivingPerson ... people) {
        Arrays.stream(people).forEach(executorService::submit);
    }

    public void endWorld() {
        executorService.shutdownNow();
        try {
            executorService.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS);
        } catch (InterruptedException e) {
            System.err.println("Failed to finish thread executor!");
        }
    }
}

每 LivingPerson 看起来像这样

public class LivingPerson implements IdentifiableRunnable {
    // investigate global world variable
    private boolean isRunning = true;

    private final StatefulPerson person;

    public LivingPerson(StatefulPerson person) {
        this.person = person;
    }

    @Override
    public void run() {
        System.out.println("Initial state: person=" + person.getRawPerson());
        while (isRunning) { // for now forever
            try {
                Thread.sleep(1000); // do transition every 1 seconds

                LifeState state = person.nextState();
                System.out.println(getPerson().getName() + " " + state.getActionLabel());
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
        System.out.println("End state: person=" + person.getRawPerson());
    }

    @Override
    public String getId() {
        Person person = getPerson();
        return person.getId() + " - " + person.getName();
    }

    @Override
    public void terminate() {
        isRunning = false;
    }

    // returns clone instead of existing
    private Person getPerson() {
        return person.getRawPerson();
    }

}

我想用人名和唯一标识符来命名每个线程。

IdentifiableRunnable

public interface IdentifiableRunnable extends Runnable {
    String getId();
    void terminate();
}

我像这样初始化一切

WorldLifecycle world = new WorldLifecycle();
LivingPerson male = createPerson("John", 40, Gender.MALE);
LivingPerson female = createPerson("Helen", 25, Gender.FEMALE);

System.out.println("Starting world...");
world.startWorld();

world.bringLife(male, female);

// just wait
try {
    Thread.sleep(10000);
} catch (InterruptedException e) {
    throw new RuntimeException(e);
}

System.out.println("Destroying world...");
world.endWorld();

但是当我试着运行它时,我得到了错误

Exception in thread "main" java.lang.ClassCastException: java.util.concurrent.ThreadPoolExecutor$Worker cannot be cast to com.lapots.breed.lifecycle.api.Ide
ntifiableRunnable
at com.lapots.breed.lifecycle.WorldLifecycle.lambda$startWorld$0(WorldLifecycle.java:14)
at java.util.concurrent.ThreadPoolExecutor$Worker.<init>(ThreadPoolExecutor.java:612)
at java.util.concurrent.ThreadPoolExecutor.addWorker(ThreadPoolExecutor.java:925)
at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1357)
at java.util.concurrent.AbstractExecutorService.submit(AbstractExecutorService.java:112)
at java.util.Spliterators$ArraySpliterator.forEachRemaining(Spliterators.java:948)
at java.util.stream.ReferencePipeline$Head.forEach(ReferencePipeline.java:580)
at com.lapots.breed.lifecycle.WorldLifecycle.bringLife(WorldLifecycle.java:20)
at com.lapots.breed.Sandbox.main(Sandbox.java:23)

identifiableRunnable 在 ThreadFactory .

1 回复 | 直到 6 年前

Coder-Man 6 年前

如果使用调试器跟踪库调用,则会注意到 newThread

Worker(Runnable firstTask) {
    setState(-1); // inhibit interrupts until runWorker
    this.firstTask = firstTask;
    this.thread = getThreadFactory().newThread(this);
}

所以它被传递给 Worker IdentifiableRunnable .

要并行执行流,应该使用 ForkJoinPool 如何在这篇文章中描述,在第一个答案 Custom thread pool in Java 8 parallel stream

顺便说一句,下面是如何压缩帖子中的代码:

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.stream.Stream;

public class Main {

    interface TestInterface extends Runnable{
        default String getId() {
            return "2";
        }
    }
    static class TestClass implements TestInterface {

        @Override
        public void run() {
            System.out.println("TEST");
        }
    }

    public static void main(String[] args) {
        ExecutorService exec = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors(), new ThreadFactory() {
            @Override
            public Thread newThread(final Runnable r) {
                String id = ((TestInterface)r).getId();
                Thread t = new Thread(r, id);
                t.setDaemon(true);
                return t;
            }
        });

        Stream.of(new TestClass(),new TestClass(),new TestClass(),new TestClass()).forEach(exec::submit);

    }
}