在Jgroup中使用自定义线程池和执行器

Question

如果我们查看Jgroup的DefaultThreadFactory，就会看到下面的代码

protected Thread newThread(Runnable r,String name,String addr,String cluster_name) { String thread_name=getNewThreadName(name, addr, cluster_name); Thread retval=new Thread(r, thread_name); retval.setDaemon(createDaemons); return retval; }

由于使用了新线程，所以我相信在托管服务器环境中，这可能会导致问题，也不是一个好的实践。

如果我只是将默认的线程工厂和执行器替换为WebSphere的托管工厂和执行器，那么Jgroup的行为是否仍然相同？

任何指示都会有帮助..？

更新

我的意图是使用JGroups和WebSphere作为8.5。我渴望没有任何未管理的线程。我的主要用例是领导人选举和一些信息传递。它将用于管理，并确保集群中只运行一个poller。

WAS8.5仍然使用CommonJ api进行工作管理。

我正在使用Spring抽象任务执行器和调度程序。

最初很容易用任务执行器代替ThreadPools，因为它们共享Executor。

必须对TaskScheduler进行调整，以便与您的TimeScheduler接口协同工作。它们非常相似，从ScheduledExecutorService扩展可能是一种选择。我实现了您的接口，并将其委托给Springs。

主要问题在于ThreadFactory。CommonJ没有这个概念。为此，我创建了一个ThreadWrapper，它封装了Runnable，并在调用"Thread's“start方法时将其委托给TaskExecutor。我忽略了线程重命名功能，因为这不会产生任何影响。

public Thread newThread(Runnable command) {
    log.debug("newThread");
    RunnableWrapper wrappedCommand = new RunnableWrapper(command);
    return new ThreadWrapper(taskExecutor, wrappedCommand);
}

public synchronized void start() {
    try { 
        taskExecutor.execute(runnableWrapper);          
    } catch (Exception e) {
        throw new UnableToStartException(e);
    }
}

这就是我遇到麻烦的地方。问题在运输中。在许多情况下，在某些内部运行程序的run方法中，例如DiagnosticsHandler、TP的TransferQueueBundler和GMS的ViewHandler，都有一个there语句检查线程。

public class DiagnosticsHandler implements Runnable {
    public void run() {
        byte[] buf;
        DatagramPacket packet;
        while(Thread.currentThread().equals(thread)) {
            //...
        }
    }
}

protected class TransferQueueBundler extends BaseBundler implements Runnable {
    public void run() {
        while(Thread.currentThread() == bundler_thread) {
            //...
        }
    }
}

class ViewHandler implements Runnable {
    public void run() {
        long start_time, wait_time;  // ns
        long timeout=TimeUnit.NANOSECONDS.convert(max_bundling_time, TimeUnit.MILLISECONDS);
        List<Request> requests=new LinkedList<>();
        while(Thread.currentThread().equals(thread) && !suspended) {
            //...
        }
    }
}

这与我们的线包装不合作。如果可以更改这一点，以便在存储的线程上调用equals方法，则可以重写它。

正如您从不同的片段中可以看到的，有不同的实现和级别的保护，不同的包，保护和公共。这增加了扩展类的难度。

所有这些完成后，它仍然没有完全消除非托管线程的问题。

我使用了属性文件方法来创建协议栈。一旦设置了属性，就会初始化协议堆栈。若要删除由底层协议创建的计时器线程，请执行以下操作。必须在初始化要初始化的堆栈之前设置TimeScheduler。

一旦完成，所有线程都将被管理。

对于如何更容易地实现这一点，你有什么建议吗？

Answer 1

是的，您可以在线程池中注入，详细信息请参见1。

1