java - 使用 ThreadPoolExecutor 和 DiscardPolicy

Question

我需要使用 ThreadPoolExecutor 创建一个客户端队列，并能够在客户端超过某个数量(例如 5)时删除客户端。这是一种 DDOS 保护。当客户端 #6 请求我的服务器时 - 它被丢弃，等等。我得到了服务器和客户端代码，但我不知道如何实现 ThreadPoolExecutor 和 DiscardPolicy。想法或例子？

简单服务器:

   import java.io.IOException;
    import java.io.InputStream;
    import java.io.ObjectInputStream;
    import java.io.OutputStream;
    import java.net.ServerSocket;
    import java.net.Socket;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.Semaphore;
    import java.util.logging.Level;
    import java.util.logging.Logger;

    public class Server {

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {

        ServerSocket server = new ServerSocket (3000);

        ExecutorService es = Executors.newFixedThreadPool(2);

        Semaphore semaphore = new Semaphore (2);

        while(true){

        semaphore.acquire();

        Socket accept2 = server.accept();

        es.execute(()->{
            try (Socket accept = accept2) {
            serve(accept);
            } catch (Exception exception) {
                Logger.getLogger(Server.class.getName()).log(Level.SEVERE, null, exception);
            } 
            finally {
                semaphore.release();

            }
        });        

        }

    }

    private static void serve(final Socket accept) throws ClassNotFoundException, IOException {
        InputStream inputStream = accept.getInputStream();
        OutputStream outputStream = accept.getOutputStream();

        ObjectInputStream inputStream2 = new ObjectInputStream (inputStream);

        while (true){
            Object readObject = inputStream2.readObject();
            System.out.println(readObject);
        }

        }

    }

还有一个简单的客户端:

  import java.io.IOException;
    import java.io.ObjectOutputStream;
    import java.net.Socket;

    public class Client {

    public static void main(String[] args) throws IOException, InterruptedException {
        Socket socket = new Socket ("localhost", 3000);
        ObjectOutputStream oos = new ObjectOutputStream (
                socket.getOutputStream());
        oos.writeObject("First!");
        Thread.sleep(10000);
        oos.writeObject("First again!");
        Thread.sleep(10000);
        oos.writeObject("First again again!");

        }

    }

Answer 1

将ThreadPoolExecutor与DiscardPolicy一起使用，如下所示:

  int poolSize=1;
  int maxPoolSize=2;
  int queueSize=5;
  long aliveTive=1000;
  ArrayBlockingQueue<Runnable> queue= new ArrayBlockingQueue<Runnable>(queueSize);
  ThreadPoolExecutor executor= new ThreadPoolExecutor(poolSize,maxPoolSize,aliveTive,
                    TimeUnit.MILLISECONDS,queue,new ThreadPoolExecutor.DiscardPolicy());
}

拒绝的任务:

New tasks submitted in method execute(Runnable) will be rejected when the Executor has been shut down, and also when the Executor uses finite bounds for both maximum threads and work queue capacity, and is saturated.

无论哪种情况，execute 方法都会调用其 RejectedExecutionHandler 的 RejectedExecutionHandler.rejectedExecution(Runnable, ThreadPoolExecutor) 方法。

提供了四种预定义处理程序策略:

1.ThreadPoolExecutor.AbortPolicy2.ThreadPoolExecutor.CallerRunsPolicy3.ThreadPoolExecutor.DiscardPolicy4.ThreadPoolExecutor.DiscardOldestPolicy

看看这个 documentation页面了解更多详情

如果以上四个不能满足您的目的，您可以通过扩展 RejectedExecutionHandler 来编写自定义 RejectionHandler。

当我需要知道哪个任务被拒绝并将这些失败保留在磁盘或数据库中时，我使用了自定义处理程序。